RACSignal分析
本文将分析冷信号RACSignal和热信号RACSubject的实现原理。
下面是美团的臧成威老师的三篇文章
- 首先，来说下冷热信号的概念。冷信号是指被动的、只有当你订阅时才会给订阅者发送消息切每次发送消息都会重新开始。
在RAC中冷信号是指RACSignal，该信号在创建时保存了didSubscribe代码块，在每次有订阅者订阅当前信号时，都会执行一遍向订阅者发送消息。
```
RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@1];
[subscriber sendNext:@2];
[subscriber sendNext:@3];
[subscriber sendError:[NSError errorWithDomain:@"失败了" code:1 userInfo:nil]];
[subscriber sendCompleted];
return nil;
}];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"%@",x);
} error:^(NSError * _Nullable error) {
NSLog(@"%@",error.domain);
} completed:^{
NSLog(@"结束");
}];
打印：
```
2019-03-20 16:52:50.103174+0800 DiamondPark[38331:1657597] Subscriber 1 recveive: 2
2019-03-20 16:52:51.103146+0800 DiamondPark[38331:1657597] Subscriber 1 recveive: 3
2019-03-20 16:52:51.103295+0800 DiamondPark[38331:1657597] Subscriber 2 recveive: 3
注意：error和completed方法只取前面的一个。
2.接下来说说热信号，RAC热信号是指RACSubject，他可以一对多，信号可以于订阅者共享信息，而且是主动的时刻发送。
RACSubject *subject = [RACSubject subject];
[[RACScheduler mainThreadScheduler] afterDelay:1 schedule:^{
[subject sendNext:@(1)];
}];
[[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
[subject sendNext:@(2)];
}];
[[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
[subject sendNext:@(3)];
}];
[[RACScheduler mainThreadScheduler] afterDelay:1.1 schedule:^{
[subject subscribeNext:^(id _Nullable x) {
NSLog(@"Subscriber 1 recveive: %@", x);
}];
}];
[[RACScheduler mainThreadScheduler] afterDelay:2.1 schedule:^{
[subject subscribeNext:^(id _Nullable x) {
NSLog(@"Subscriber 2 recveive: %@", x);
}];
}];
打印：
2019-03-20 16:54:37.792243+0800 DiamondPark[38415:1665726] Subscriber 1 recveive: 1
2019-03-20 16:54:37.792437+0800 DiamondPark[38415:1665726] Subscriber 1 recveive: 2
2019-03-20 16:54:37.792524+0800 DiamondPark[38415:1665726] Subscriber 1 recveive: 3
2019-03-20 16:54:38.684566+0800 DiamondPark[38415:1665726] Subscriber 2 recveive: 1
2019-03-20 16:54:38.684704+0800 DiamondPark[38415:1665726] Subscriber 2 recveive: 2
2019-03-20 16:54:38.684831+0800 DiamondPark[38415:1665726] Subscriber 2 recveive: 3
热信号是主动的，即使你没有订阅事件，它仍然会时刻推送。而冷信号是被动的，只有当你订阅的时候，它才会发送消息。
热信号可以有多个订阅者，是一对多，信号可以与订阅者共享信息。而冷信号只能一对一，当有不同的订阅者，消息会从新完整发送。
RACStream
RACStream是RACSignal和RACSequence父类，定义了一些基本的方法。
- (__kindof RACStream<ValueType> *)empty; (用来返回一个空值)
- (__kindof RACStream<ValueType> *)return:(nullable ValueType)value;（把一个值包装秤对应的RACStream的子类型)
- (__kindof RACStream *)bind:(RACStreamBindBlock (^)(void))block;（改变当前流对象的方法)
- (__kindof RACStream *)concat:(RACStream *)stream;（用来连接信号，子类实现具体如何连接）
- (__kindof RACStream *)zipWith:(RACStream *)stream;(压缩两个信号，子类实现具体如何压缩。)
其中有个分类Operations定义了许多方便的方法，实现以上几个基础方法便可直接使用Operations中的方法。
一、flattenMap：返回经过bind改变的之后的对象让block进行包装之后返回改变过流的信号
- (__kindof RACStream *)flattenMap:(__kindof RACStream * (^)(id value))block {
Class class = self.class;
return [[self bind:^{
return ^(id value, BOOL *stop) {
id stream = block(value) ?: [class empty];
NSCAssert([stream isKindOfClass:RACStream.class], @"Value returned from -flattenMap: is not a stream: %@", stream);
return stream;
};
}] setNameWithFormat:@"[%@] -flattenMap:", self.name];
}
如在RACSignal中(以下都以RACSignal为例)：
RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@"111"];
return nil;
}];
[[signal flattenMap:^__kindof RACSignal * _Nullable(id _Nullable value) {
return [RACSignal return:@([value length])];
}]subscribeNext:^(id _Nullable x) {
NSLog(@"%@",x);
}] ;
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"mmmmm ::%@",x);
}];
打印：
2019-03-20 19:41:06.435679+0800 DiamondPark[53605:2257423] 3
2019-03-20 19:41:06.435909+0800 DiamondPark[53605:2257423] mmmmm ::111
二、flatten不做任何处理，只是创建了一个新的信号，并返回一个值，是为了处理merge方法而实现的
- (__kindof RACStream *)flatten {
return [[self flattenMap:^(id value) {
return value;
}] setNameWithFormat:@"[%@] -flatten", self.name];
}
三、map只用返回一个value
- (__kindof RACStream *)map:(id (^)(id value))block {
NSCParameterAssert(block != nil);
Class class = self.class;
return [[self flattenMap:^(id value) {
return [class return:block(value)];
}] setNameWithFormat:@"[%@] -map:", self.name];
}
四、mapReplace，调用了map方法，不管收到什么信号都返回repleace的值
- (__kindof RACStream *)mapReplace:(id)object {
return [[self map:^(id _) {
return object;
}] setNameWithFormat:@"[%@] -mapReplace: %@", self.name, RACDescription(object)];
}
五、filter，如果返回值为Yes则通知订阅者，否者，无效果，用来过滤消息
- (__kindof RACStream *)filter:(BOOL (^)(id value))block {
NSCParameterAssert(block != nil);
Class class = self.class;
return [[self flattenMap:^ id (id value) {
if (block(value)) {
return [class return:value];
} else {
return class.empty;
}
}] setNameWithFormat:@"[%@] -filter:", self.name];
}
六、ignore 忽略某些值
- (__kindof RACStream *)ignore:(id)value {
return [[self filter:^ BOOL (id innerValue) {
return innerValue != value && ![innerValue isEqual:value];
}] setNameWithFormat:@"[%@] -ignore: %@", self.name, RACDescription(value)];
}
七、reduceEach
RACBlockTrampoline是一个实现动态调用方法的类。他根据需要传入一个block和一个tuple对象来实现block变参调用
- (__kindof RACStream *)reduceEach:(RACReduceBlock)reduceBlock {
NSCParameterAssert(reduceBlock != nil);
__weak RACStream *stream __attribute__((unused)) = self;
return [[self map:^(RACTuple *t) {
NSCAssert([t isKindOfClass:RACTuple.class], @"Value from stream %@ is not a tuple: %@", stream, t);
return [RACBlockTrampoline invokeBlock:reduceBlock withArguments:t];
}] setNameWithFormat:@"[%@] -reduceEach:", self.name];
}
使用如下：
RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:RACTuplePack(@"11",@"22",@"333")];
// [subscriber sendNext:@"222"];
return nil;
}];
[[signal reduceEach:^id _Nonnull(id a, id b, id c){
return @([a integerValue] + [b integerValue] + [c integerValue]);
}] subscribeNext:^(id _Nullable x) {
NSLog(@"%@",x);
} completed:^{
NSLog(@"空值");
}] ;
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"mmmmm ::%@",x);
}];
它要求发送的信号为RACTuple对象。reduceEach传入的对象为block，在这个block方法中，调用了map方法，map返回的是一个RACBlockTrampoline处理过的id对象，
[RACBlockTrampoline invokeBlock:reduceBlock withArguments:t];
经过RACBlockTrampoline会根据传入的元组对象来调用相对应的多参block方法处理。
这里要说一下：
id block = ^(id a, id b){
NSLog(@"%@%@",a,b);
};
id (^c )(id , id , id) = ^ (id a, id b, id c){
return @([a integerValue] + [b integerValue] + [c integerValue]);
};
c = block;
c(@1,@2,@3);
这个block一共可以设置三个参数，因为c的定义，在第三部中，将c的地址志祥block，虽然可以传递三个参数，但是在block实现中就可以动态使用其中的一个或多个参数
skip，跳过前几次的信号
- (__kindof RACStream *)skip:(NSUInteger)skipCount {
Class class = self.class;
return [[self bind:^{
__block NSUInteger skipped = 0;
return ^(id value, BOOL *stop) {
if (skipped >= skipCount) return [class return:value];
skipped++;
return class.empty;
};
}] setNameWithFormat:@"[%@] -skip: %lu", self.name, (unsigned long)skipCount];
}
take，取前几次的信号，次数达到后会调用complete方法
- (__kindof RACStream *)take:(NSUInteger)count {
Class class = self.class;
if (count == 0) return class.empty;
return [[self bind:^{
__block NSUInteger taken = 0;
return ^ id (id value, BOOL *stop) {
if (taken < count) {
++taken;
if (taken == count) *stop = YES;
return [class return:value];
} else {
return nil;
}
};
}] setNameWithFormat:@"[%@] -take: %lu", self.name, (unsigned long)count];
}
join:block:
因为传入的是一个数组，用for循环使用map方法生成一个信号并调用block，返回当前的信号和新生成的信号，并将老的信号复制给当前信号
for (RACStream *stream in streams) {
if (current == nil) {
current = [stream map:^(id x) {
return RACTuplePack(x);
}];
continue;
}
current = block(current, stream);
}
if (current == nil) return [self empty];
根据上面的算法得到的ractuple都为两个元素，因为zip的方法，所以结构类似这样：
(((0),1）,2)
现在需要把它转化为一个正常的元组对象，即（1，2，3，4，5）
if (current == nil) return [self empty];
return [current map:^(RACTuple *xs) {
NSMutableArray *values = [[NSMutableArray alloc] init];
while (xs != nil) {
[values insertObject:xs.last ?: RACTupleNil.tupleNil atIndex:0];
xs = (xs.count > 1 ? xs.first : nil);
}
return [RACTuple tupleWithObjectsFromArray:values];
}];
for循环取第一个然后将后面一个插入到数组中，如果第一个是nil或者数组的个数变为1，则返回nil跳出循环，并用新的数组生成RACTuple对象；
zip，讲一个list压缩
return [[self join:streams block:^(RACStream *left, RACStream *right) {
return [left zipWith:right];
}] setNameWithFormat:@"+zip: %@", streams];
zip：reduce：
+ (__kindof RACStream *)zip:(id<NSFastEnumeration>)streams reduce:(RACGenericReduceBlock)reduceBlock {
NSCParameterAssert(reduceBlock != nil);
RACStream *result = [self zip:streams];
// Although we assert this condition above, older versions of this method
// supported this argument being nil. Avoid crashing Release builds of
// apps that depended on that.
if (reduceBlock != nil) result = [result reduceEach:reduceBlock];
return [result setNameWithFormat:@"+zip: %@ reduce:", streams];
}
去zip的一个或多个元素
concat，将一个list合并
+ (__kindof RACStream *)concat:(id<NSFastEnumeration>)streams {
RACStream *result = self.empty;
for (RACStream *stream in streams) {
result = [result concat:stream];
}
return [result setNameWithFormat:@"+concat: %@", streams];
}
scanWithStart：reduce：reduceWithIndex：，设置一个初始值然后根据发送的信号来返回数据，是叠加的
- (__kindof RACStream *)scanWithStart:(id)startingValue reduce:(id (^)(id running, id next))reduceBlock {
NSCParameterAssert(reduceBlock != nil);
return [[self
scanWithStart:startingValue
reduceWithIndex:^(id running, id next, NSUInteger index) {
return reduceBlock(running, next);
}]
setNameWithFormat:@"[%@] -scanWithStart: %@ reduce:", self.name, RACDescription(startingValue)];
}
scanWithStart：reduceWithIndex：，设置一个初始值然后根据发送的信号和index来返回数据，是叠加的
- (__kindof RACStream *)scanWithStart:(id)startingValue reduceWithIndex:(id (^)(id, id, NSUInteger))reduceBlock {
NSCParameterAssert(reduceBlock != nil);
Class class = self.class;
return [[self bind:^{
__block id running = startingValue;
__block NSUInteger index = 0;
return ^(id value, BOOL *stop) {
running = reduceBlock(running, value, index++);
return [class return:running];
};
}] setNameWithFormat:@"[%@] -scanWithStart: %@ reduceWithIndex:", self.name, RACDescription(startingValue)];
}
combinePreviousWithStart：reduce：f(x) = x(x-1)+x(x);f(0) = x(0)+a,即，第一次的时候使用startvalue和第一个收到值，剩余的使用本次收到的值和上一次的值
- (__kindof RACStream *)combinePreviousWithStart:(id)start reduce:(id (^)(id previous, id next))reduceBlock {
NSCParameterAssert(reduceBlock != NULL);
return [[[self
scanWithStart:RACTuplePack(start)
reduce:^(RACTuple *previousTuple, id next) {
id value = reduceBlock(previousTuple[0], next);
return RACTuplePack(next, value);
}]
map:^(RACTuple *tuple) {
return tuple[1];
}]
setNameWithFormat:@"[%@] -combinePreviousWithStart: %@ reduce:", self.name, RACDescription(start)];
}
takeUntilBlock，取一定次数，返回一个判断条件，次数达到后会调用complete方法
疑问：为什么返回nil后之后的方法就不调用了？销毁了吗
- (__kindof RACStream *)takeUntilBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
Class class = self.class;
return [[self bind:^{
return ^ id (id value, BOOL *stop) {
if (predicate(value)) return nil;
return [class return:value];
};
}] setNameWithFormat:@"[%@] -takeUntilBlock:", self.name];
}
takeWhileBlock，takeUntilBlock的取反
- (__kindof RACStream *)takeWhileBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
return [[self takeUntilBlock:^ BOOL (id x) {
return !predicate(x);
}] setNameWithFormat:@"[%@] -takeWhileBlock:", self.name];
}
skipUntilBlock,只要满足条件，就不跳过
- (__kindof RACStream *)skipUntilBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
Class class = self.class;
return [[self bind:^{
__block BOOL skipping = YES;
return ^ id (id value, BOOL *stop) {
if (skipping) {
if (predicate(value)) {
skipping = NO;
} else {
return class.empty;
}
}
return [class return:value];
};
}] setNameWithFormat:@"[%@] -skipUntilBlock:", self.name];
}
skipWhileBlock，只要不满足条件，就不跳过
- (__kindof RACStream *)skipWhileBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
return [[self skipUntilBlock:^ BOOL (id x) {
return !predicate(x);
}] setNameWithFormat:@"[%@] -skipWhileBlock:", self.name];
}
distinctUntilChanged，如果前面的获取的值和后面获取的值不相同，就输送值
- (__kindof RACStream *)distinctUntilChanged {
Class class = self.class;
return [[self bind:^{
__block id lastValue = nil;
__block BOOL initial = YES;
return ^(id x, BOOL *stop) {
if (!initial && (lastValue == x || [x isEqual:lastValue])) return [class empty];
initial = NO;
lastValue = x;
return [class return:x];
};
}] setNameWithFormat:@"[%@] -distinctUntilChanged", self.name];
}
RACSignal实现分析
RACSignal是RAC的基石，接下来我们（wo）就来分析一下RACSignal的实现原理。
RACSignal有6个子类：
- RACChannelTerminal,通道终端，代表RACChannel的一个终端，用来实现双向绑定。
- RACDynamicSignal:动态信号，使用一个 block - 来实现订阅行为，我们在使用 RACSignal 的 +createSignal: 方法时创建的就是该类的实例
- RACEmptySignal，空信号，用来实现 RACSignal 的 +empty 方法
- RACErrorSignal，错误信号，用来实现 RACSignal 的 +error: 方法；
- RACReturnSignal，一元信号，用来实现 RACSignal 的 +return: 方法；
- RACSubject，热信号，既可以是发送者也可以是接受者
RACSignal分析
冷信号的原理其实是：在create方法中保存的一个block，当有订阅者调用（subscribeNext:)
方法时首先会创建一个RACSubscriber对象，并将3个block保存,然后调用（subscribe：）方法， 创建一个透传发送消息的对象实现消息的转发，在方法中调用创建时的block，并传回发送者，所以我们可以在createblock里面设置预留的消息队列，并且在每次更新订阅者的时候，都会走这个block；
RACSignal的头文件一共有几个主要方法：
二、自身方法
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
return [RACDynamicSignal createSignal:didSubscribe];
}
其实是根据block创建RACDynamicSignal对象。
+ (RACSignal *)error:(NSError *)error
+ (RACSignal *)error:(NSError *)error {
return [RACErrorSignal error:error];
}
创建一个RACErrorSignal对象
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock
+ (RACSignal *)return:(id)value
+ (RACSignal *)return:(id)value {
return [RACReturnSignal return:value];
}
创建一个RACReturnSignal对象
+ (RACSignal *)startLazilyWithScheduler:(RACScheduler *)scheduler block:(void (^)(id<RACSubscriber> subscriber))block
+ (RACSignal *)startEagerlyWithScheduler:(RACScheduler *)scheduler block:(void (^)(id<RACSubscriber> subscriber))block
+ (RACSignal *)never
+ (RACSignal *)never {
return [[self createSignal:^ RACDisposable * (id<RACSubscriber> subscriber) {
return nil;
}] setNameWithFormat:@"+never"];
}
创建一个没有初始化任何消息的信号
+ (RACSignal *)empty
+ (RACSignal *)empty {
return [RACEmptySignal empty];
}
创建一个RACEmptySignal
二、实现父类方法
bind实现
bind的实现比较长我们可以分步分析，在这个方法中创建一个新的信号量。，首先completeSignal这个block，
__block volatile int32_t signalCount = 1;
void (^completeSignal)(RACDisposable *) = ^(RACDisposable *finishedDisposable) {
if (OSAtomicDecrement32Barrier(&signalCount) == 0) {
[subscriber sendCompleted];
[compoundDisposable dispose];
} else {
[compoundDisposable removeDisposable:finishedDisposable];
}
};
这个闭包中显示判断这个信号数量减一后是否为0，如果是0则发送完成信息，执行销毁任务，负责只是移除这个任务。
RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
void (^addSignal)(RACSignal *) = ^(RACSignal *signal) {
OSAtomicIncrement32Barrier(&signalCount);
RACSerialDisposable *selfDisposable = [[RACSerialDisposable alloc] init];
[compoundDisposable addDisposable:selfDisposable];
RACDisposable *disposable = [signal subscribeNext:^(id x) {
[subscriber sendNext:x];
} error:^(NSError *error) {
[compoundDisposable dispose];
[subscriber sendError:error];
} completed:^{
@autoreleasepool {
completeSignal(selfDisposable);
}
}];
selfDisposable.disposable = disposable;
};
在这个闭包中，将signalCount加1，创建了一个同步销毁任务，添加到聚合任务中；当前信号订阅传入的信号，因为传入的信号都为RACReturnSignal信号，在有订阅者订阅他的时候会主动调用一次sendNext:value值，在接受到这个信号后，新创建的信号也会发信息给自己的订阅者。
注意：如果发送send信号则会通过转发signal走返回值得转发，否则直接让新创建的signal进行处理，每次走block都会创建一个新的转发signal
RACSerialDisposable *selfDisposable = [[RACSerialDisposable alloc] init];
[compoundDisposable addDisposable:selfDisposable];
RACDisposable *bindingDisposable = [self subscribeNext:^(id x) {
// Manually check disposal to handle synchronous errors.
if (compoundDisposable.disposed) return;
BOOL stop = NO;
id signal = bindingBlock(x, &stop);
@autoreleasepool {
if (signal != nil) addSignal(signal);
if (signal == nil || stop) {
[selfDisposable dispose];
completeSignal(selfDisposable);
}
}
} error:^(NSError *error) {
[compoundDisposable dispose];
[subscriber sendError:error];
} completed:^{
@autoreleasepool {
completeSignal(selfDisposable);
}
}];
selfDisposable.disposable = bindingDisposable;
这个方法中，先是订阅旧的信号，然后在接受到消息后，先是判断新的信号是否已经销毁，如果销毁
则返回，如果没被销毁，拿到block返回的信号，如果信号不为nil，执行addSignalblock，否则就执行销毁任务。
在这个过程中一共有3个信号，第一个是需要转换的信号（self），第二个是转换后的信号（return），第三个是中转信号（id signal = bindingBlock(x, &stop);）这里面很重要的问题应该是循环引用吧。
concat, 合并两个信号，并顺序执行，先执行被合并的信号方法，再执行合并的信号方法
- (RACSignal *)concat:(RACSignal *)signal {
return [[RACSignal createSignal:^(id<RACSubscriber> subscriber) {
RACCompoundDisposable *compoundDisposable = [[RACCompoundDisposable alloc] init];
RACDisposable *sourceDisposable = [self subscribeNext:^(id x) {
[subscriber sendNext:x];
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
RACDisposable *concattedDisposable = [signal subscribe:subscriber];
[compoundDisposable addDisposable:concattedDisposable];
}];
[compoundDisposable addDisposable:sourceDisposable];
return compoundDisposable;
}] setNameWithFormat:@"[%@] -concat: %@", self.name, signal];
}
合并信号，即将两个信号合并到一起，在这个实现中，首先返回一个新创建的信号，在这个信号的didisubsciblock中，先是订阅一下当前信号，当当前信号完成后，再将传入的信号的发送者设置为新创新创建信号的发送者，注意在这里面还会重新执行一遍signal的didSubscribe。
在这个过程中会有三个信号，一个是执行合并的信号，将要合并的信号，以及转换后的信号，即新信号。发送消息的过程是，首先，当前的信号会发送给新的洗好，发送结束后，被合并的账号会发送消息给新信号。
zipWith实现， 合并两个信号，当两个信号都发送过一次消息后才发送,返回一个元组
__block BOOL selfCompleted = NO;
NSMutableArray *selfValues = [NSMutableArray array];
__block BOOL otherCompleted = NO;
NSMutableArray *otherValues = [NSMutableArray array];
创建两个数组，和两个bool值
void (^sendCompletedIfNecessary)(void) = ^{
@synchronized (selfValues) {
BOOL selfEmpty = (selfCompleted && selfValues.count == 0);
BOOL otherEmpty = (otherCompleted && otherValues.count == 0);
if (selfEmpty || otherEmpty) [subscriber sendCompleted];
}
};
当这两个bool值和数组有一对是完成的话，就向订阅者发送已完成的信号。
void (^sendNext)(void) = ^{
@synchronized (selfValues) {
if (selfValues.count == 0) return;
if (otherValues.count == 0) return;
RACTuple *tuple = RACTuplePack(selfValues[0], otherValues[0]);
[selfValues removeObjectAtIndex:0];
[otherValues removeObjectAtIndex:0];
[subscriber sendNext:tuple];
sendCompletedIfNecessary();
}
};
如果两个数组有一个为空数组就返回，否则，发送一个元组给订阅者，并移除。然后检查是否完成。
RACDisposable *selfDisposable = [self subscribeNext:^(id x) {
@synchronized (selfValues) {
[selfValues addObject:x ?: RACTupleNil.tupleNil];
sendNext();
}
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
@synchronized (selfValues) {
selfCompleted = YES;
sendCompletedIfNecessary();
}
}];
RACDisposable *otherDisposable = [signal subscribeNext:^(id x) {
@synchronized (selfValues) {
[otherValues addObject:x ?: RACTupleNil.tupleNil];
sendNext();
}
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
@synchronized (selfValues) {
otherCompleted = YES;
sendCompletedIfNecessary();
}
}];
订阅两个信号，处理
RACSubject
RACSubject实现了rac热信号的功能。
它子类有：
- RACBehaviorSubject（重演最后值的信号，当被订阅时，会向订阅者发送它最后接收到的值；）
- RACReplaySubject（重演信号，保存发送过的值，当被订阅时，会向订阅者重新发送这些值。）
- RACGroupedSignal（分组信号，用来实现 RACSignal 的分组功能）
它的初始化方法是：
+ (instancetype)subject;
它遵守并实现了RACSubscriber协议，所以它同时拥有接受信息和发送信息的能力。
下面讲一下实现：
- (instancetype)init
- (instancetype)init {
self = [super init];
if (self == nil) return nil;
_disposable = [RACCompoundDisposable compoundDisposable];
_subscribers = [[NSMutableArray alloc] initWithCapacity:1];
return self;
}
在初始化方法中创建了一个默认存储空间为一的可变数组。
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
NSCParameterAssert(subscriber != nil);
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
NSMutableArray *subscribers = self.subscribers;
@synchronized (subscribers) {
[subscribers addObject:subscriber];
}
[disposable addDisposable:[RACDisposable disposableWithBlock:^{
@synchronized (subscribers) {
// 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];
}
}]];
return disposable;
}
在设置发送的方法里面，将发送者存入上面创建的数组中。
并在销毁订阅者的时候移除发送者。
- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block
- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {
NSArray *subscribers;
@synchronized (self.subscribers) {
subscribers = [self.subscribers copy];
}
for (id<RACSubscriber> subscriber in subscribers) {
block(subscriber);
}
}
遍历发送者并执行block任务。
RACSubscriber代理方法
遍历调用
实现这些方法后，在添加订阅者的时候就会保存进racsubject，在发送信息的时候，使订阅者的消息发送者都发送这些信息实现实时发送。
热信号与冷信号的转换
第一种，广播：
- (void)hotCodeSignal{
RACSignal *coldSignal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
NSLog(@"Cold signal be subscribed.");
[[RACScheduler mainThreadScheduler] afterDelay:1.5 schedule:^{
[subscriber sendNext:@"A"];
}];
[[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
[subscriber sendNext:@"B"];
}];
[[RACScheduler mainThreadScheduler] afterDelay:5 schedule:^{
[subscriber sendCompleted];
}];
return nil;
}];
RACSubject *subject = [RACSubject subject];
NSLog(@"Subject created.");
[[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
[coldSignal subscribe:subject];
}];
[subject subscribeNext:^(id x) {
NSLog(@"Subscriber 1 recieve value:%@.", x);
}];
[[RACScheduler mainThreadScheduler] afterDelay:4 schedule:^{
[subject subscribeNext:^(id x) {
NSLog(@"Subscriber 2 recieve value:%@.", x);
}];
}];
}
将RACSigal的发送者设为RACSubject，并用RACSubjec设置订阅者。“但这种方式并不能终止对coldSignal的订阅”这句话的意思大概是说，冷信号的disposable并没有包含热信号的disposable,所以在销毁的时候并没有执行热信号销毁订阅的方法。
第二种，借助RACSignal的方法：
- (RACMulticastConnection *)publish;
- (RACMulticastConnection *)multicast:(RACSubject *)subject;
- (RACSignal *)replay;
- (RACSignal *)replayLast;
- (RACSignal *)replayLazily;
其中最重要的事mlticast的方法，其他方法都是借助它实现的。
RACMulticastConnection
多连接，是用来实现冷信号转热信号的。先来看头文件：
@interface RACMulticastConnection<__covariant ValueType> : NSObject
//其实传的是一个RACSubject类
@property (nonatomic, strong, readonly) RACSignal<ValueType> *signal;
//热信号绑定冷信号
- (RACDisposable *)connect;
//自动连接
- (RACSignal<ValueType> *)autoconnect RAC_WARN_UNUSED_RESULT;
@end
接下来看.m:
- (instancetype)initWithSourceSignal:(RACSignal *)source subject:(RACSubject *)subject {
NSCParameterAssert(source != nil);
NSCParameterAssert(subject != nil);
self = [super init];
_sourceSignal = source;
_serialDisposable = [[RACSerialDisposable alloc] init];
_signal = subject;
return self;
}
初始赋值方法。
- (RACDisposable *)connect {
BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);
if (shouldConnect) {
self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
}
return self.serialDisposable;
}
热信号订阅冷信号。
- (RACSignal *)autoconnect {
__block volatile int32_t subscriberCount = 0;
return [[RACSignal
createSignal:^(id<RACSubscriber> subscriber) {
OSAtomicIncrement32Barrier(&subscriberCount);
RACDisposable *subscriptionDisposable = [self.signal subscribe:subscriber];
RACDisposable *connectionDisposable = [self connect];
return [RACDisposable disposableWithBlock:^{
[subscriptionDisposable dispose];
if (OSAtomicDecrement32Barrier(&subscriberCount) == 0) {
[connectionDisposable dispose];
}
}];
}]
setNameWithFormat:@"[%@] -autoconnect", self.signal.name];
}
冷信号的信息发送者订阅热信号。
RACSubject *subject = [RACSubject subject];
RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
NSLog(@"Cold signal be subscribed.");
[[RACScheduler mainThreadScheduler] afterDelay:1.5 schedule:^{
[subscriber sendNext:@"A"];
}];
[[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
[subscriber sendNext:@"B"];
}];
[[RACScheduler mainThreadScheduler] afterDelay:5.5 schedule:^{
[subscriber sendNext:@"C"];
}];
return [RACDisposable disposableWithBlock:^{
NSLog(@"完成");
}];
}];
//用法一、
RACMulticastConnection *connet = [signal multicast:subject];
[connet connect];
// [signal subscribe:subject];
RACSignal *hot = connet.signal;
[[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
[hot subscribeNext:^(id x) {
NSLog(@"Subscribe 1 recieve value:%@.", x);
}];
}];
//用法二、
RACMulticastConnection *connet = [signal multicast:subject];
RACSignal *hot = connet.autoconnect;
[[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
[hot subscribeNext:^(id x) {
NSLog(@"Subscribe 1 recieve value:%@.", x);
}];
}];
网友评论