ReactiveCocoa-Signal

ReactiveCocoa是一个将函数响应式编程（Functional Reactive Programming）带入到Objective-C中的开源库。

RACSignal

RAC中的核心是Signal，对应的类是RACSignal。它代表一个信号，Signal会给它的订阅者（Subscriber）发送一连串的数据，订阅者接到数据之后对这些数据进行处理。
RACSignal被订阅的过程如下：

    RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
        
        [subscriber sendNext:@1];
        [subscriber sendNext:@2];
        [subscriber sendNext:@3];
        
        [subscriber sendCompleted];
        
        return [RACDisposable disposableWithBlock:^{
            NSLog(@"signal disposable");
        }];
    }];
    
    [signal subscribeNext:^(id  _Nullable x) {
        NSLog(@"value = %@",x);
    } error:^(NSError * _Nullable error) {
        NSLog(@"error = %@",error);
    } completed:^{
        NSLog(@"completed");
    }];
  • 调用RACSignal的createSignal:方法创建一个signal
  • 调用RACSignal的subscribeNext:方法，这个方法创建一个subscriber并调用subscribe:方法
  • subscribe:方法将subscriber传入didSubscribe的Block中
  • 在didSubscribe中通过[subscriber sendNext:]来执行next block
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe];
}
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    RACDynamicSignal *signal = [[self alloc] init];
    signal->_didSubscribe = [didSubscribe copy];
    return [signal setNameWithFormat:@"+createSignal:"];
}

RACSignal的createSignal:方法通过调用子类RACDynamicSignald createSignal:方法返回一个Signal，并保存didSubscribe这个block。

- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
    NSCParameterAssert(nextBlock != NULL);
    NSCParameterAssert(errorBlock != NULL);
    NSCParameterAssert(completedBlock != NULL);
    
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
    return [self subscribe:o];
}
+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
    RACSubscriber *subscriber = [[self alloc] init];

    subscriber->_next = [next copy];
    subscriber->_error = [error copy];
    subscriber->_completed = [completed copy];

    return subscriber;
}

subscribeNext:方法首先创建一个subscriber，并把next,error,completed三个block保存到subscriber中。接着调用RACDynamicSignal的subscribe:方法。

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCParameterAssert(subscriber != nil);

    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

    if (self.didSubscribe != NULL) {
        RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];

        [disposable addDisposable:schedulingDisposable];
    }
    
    return disposable;
}

在subscribe:方法中执行保存在Signal中的didSubscribe，传入参数subscriber。

- (void)sendNext:(id)value {
    @synchronized (self) {
        void (^nextBlock)(id) = [self.next copy];
        if (nextBlock == nil) return;

        nextBlock(value);
    }
}

在didSubscribe中，传入的subscriber调用sendNext:方法执行保存在subscriber中的nextBlock。

通过上面的代码，可以发现ReactiveCocoa是通过block实现的订阅过程。signal中保存了didSubscribe，subscriber中保存了nextBlock，errorBlock，completedBlock。didSubscribe的参数为一个subscriber，在block中，传入的subscriber调用sendNext:，sendError:，sendCompleted方法执行保存在subscriber中的block。didSubscribe的执行在signal的subscribe:方法中，而RAC将subscriber的初始化和subscribe:的调用放在subscribeNext:之中。

按照上面的订阅过程，可以发现信号只有在被订阅的时候才会发送信号值，并且每订阅一次，didSubscribe就会被执行一次。这种信号称为冷信号。既然有冷信号，那么也有热信号，冷热信号的区别是：

  1. 热信号是主动的，尽管你并没有订阅事件，但是它会时刻推送，就像鼠标移动；而冷信号是被动的，只有当你订阅的时候，它才会发布消息。

  2. 热信号可以有多个订阅者，是一对多，集合可以与订阅者共享信息；而冷信号只能一对一，当有不同的订阅者，消息是重新完整发送。

RACSubject

在RAC中，冷信号基本都是RACSignal及其子类，而热信号则是RACSubject及其子类。

@interface RACSubject<ValueType> : RACSignal<ValueType> <RACSubscriber>

RACSubject继承RACSignal，并且遵守RACSubscriber协议，这说明既可以发送信号，也可以订阅信号。
RACSubject的使用如下：

    RACSubject *subject = [RACSubject subject];
    
    [subject subscribeNext:^(id x) {
        NSLog(@"subscribe %@ ", x);
    }];
    [subject sendNext:@1];
    [subject sendNext:@2];

接下来看一下RACSubject与RACSignal有什么不同

- (instancetype)init {
    self = [super init];
    if (self == nil) return nil;

    _disposable = [RACCompoundDisposable compoundDisposable];
    _subscribers = [[NSMutableArray alloc] initWithCapacity:1];
    
    return self;
}

可以看到RACSubject有一个subscribers数组

- (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;
}

通过subscribe:方法可以发现，subject的每次订阅都是将subscriber加到subscribers数组中。

- (void)sendNext:(id)value {
    [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
        [subscriber sendNext:value];
    }];
}

从sendNext:的实现可以看出，每次RACSubject对象sendNext，都会对其中保留的subscribers进行sendNext，执行subscriber保存的nextBlock。

RACSubject满足热信号的特点，它即使没有subscriber，因为自己继承了RACSubscriber协议，所以自己本身就可以发送信号。由于RACSubject将subscriber保存到数组中，所以可以给多个subscriber发送消息，并且subscriber只能接收到订阅之后发送的消息，订阅之前已经发出的消息则无法收到。

RACSubject有一个子类RACReplaySubject，RACReplaySubject相对于RACSubject增加了一个数组用来保存发送过的数据，当有新的订阅时，会重发之前发送的数据。

- (instancetype)initWithCapacity:(NSUInteger)capacity {
    self = [super init];
    
    _capacity = capacity;
    _valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
    
    return self;
}
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];

    RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
        @synchronized (self) {
            for (id value in self.valuesReceived) {
                if (compoundDisposable.disposed) return;

                [subscriber sendNext:(value == RACTupleNil.tupleNil ? nil : value)];
            }

            if (compoundDisposable.disposed) return;

            if (self.hasCompleted) {
                [subscriber sendCompleted];
            } else if (self.hasError) {
                [subscriber sendError:self.error];
            } else {
                RACDisposable *subscriptionDisposable = [super subscribe:subscriber];
                [compoundDisposable addDisposable:subscriptionDisposable];
            }
        }
    }];

    [compoundDisposable addDisposable:schedulingDisposable];

    return compoundDisposable;
}

冷信号转化为热信号

由于RACSignal是冷信号，那么在用于网络请求或者是数据计算的时候，如果有多次订阅导致多次执行didSubscribe会发生重复请求以及数据错误。为了避免这种情况，可以使用RACMulticastConnection将冷信号转化为热信号。

RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
        
        [subscriber sendNext:@1];
        [subscriber sendNext:@2];
        [subscriber sendNext:@3];
        
        [subscriber sendCompleted];
        
        return [RACDisposable disposableWithBlock:^{
            NSLog(@"signal disposable");
        }];
    }];
    
    RACMulticastConnection *connection = [signal multicast:[RACSubject subject]];
    
    [connection.signal subscribeNext:^(id  _Nullable x) {
        NSLog(@"value = %@",x);
    } error:^(NSError * _Nullable error) {
        NSLog(@"error = %@",error);
    } completed:^{
        NSLog(@"completed");
    }];
    
    [connection connect];

对一个Signal进行multicast之后，我们是对connection.signal进行订阅而不是原来的signal。为什么connection.signal是热信号，我们看一下。

- (RACMulticastConnection *)multicast:(RACSubject *)subject {
    [subject setNameWithFormat:@"[%@] -multicast: %@", self.name, subject.name];
    RACMulticastConnection *connection = [[RACMulticastConnection alloc] initWithSourceSignal:self subject:subject];
    return connection;
}
- (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;
}

通过RACMulticastConnection的初始化方法可以看出，RACMulticastConnection将原signal和传入的RACSubject保存，connection的signal实际是传入的RACSubject。

- (RACDisposable *)connect {
    BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);

    if (shouldConnect) {
        self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
    }

    return self.serialDisposable;
}

[self.sourceSignal subscribe:_signal]会执行self.sourceSignal的didSubscribe这个block，并且将_signal当做订阅者。_signal在didSubscribe中会sendNext，这里的这个signal就是[RACSubject subject]。可以看出，一旦进入到这个didSubscribe中，后续的操作都是对这个[RACReplaySubject subject]进行的，与原来的sourceSignal彻底无关了。只有调用connect时，才执行sourceSignal的didSubscribe。这样我们不管订阅多少次connection.signal，只要connect只调用一次，那么didSubscribe也就只会执行一次。

- (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];
}

autoconnect通过订阅self.signal生成的信号，允许我们在第一次订阅connection.signal的时候自动调用connect。

除了multicast:方法，还有publish，replay，replayLast，replayLazily方法

- (RACMulticastConnection *)publish {
    RACSubject *subject = [[RACSubject subject] setNameWithFormat:@"[%@] -publish", self.name];
    RACMulticastConnection *connection = [self multicast:subject];
    return connection;
}

publish方法只不过是去调用了multicast:方法，publish内部会新建好一个RACSubject，并把它当成入参传递给RACMulticastConnection。

- (RACSignal *)replay {
    RACReplaySubject *subject = [[RACReplaySubject subject] setNameWithFormat:@"[%@] -replay", self.name];

    RACMulticastConnection *connection = [self multicast:subject];
    [connection connect];

    return connection.signal;
}

replay在创建RACMulticastConnection之后马上调用了connect方法，所以使用了RACReplaySubject，因为它会保存发送过的消息，当订阅之后依然会收到之前发送的消息，如果使用RACSubject会收不到消息。

- (RACSignal *)replayLast {
    RACReplaySubject *subject = [[RACReplaySubject replaySubjectWithCapacity:1] setNameWithFormat:@"[%@] -replayLast", self.name];

    RACMulticastConnection *connection = [self multicast:subject];
    [connection connect];

    return connection.signal;
}

replayLast和replay的实现基本一样，不同的是replayLast将Capacity值设为1，意味着只保留最新的值。

- (RACSignal *)replayLazily {
    RACMulticastConnection *connection = [self multicast:[RACReplaySubject subject]];
    return [[RACSignal
        defer:^{
            [connection connect];
            return connection.signal;
        }]
        setNameWithFormat:@"[%@] -replayLazily", self.name];
}

replayLazily将connect放在了defer中，当订阅信号的时候它才会被调用，如果含有和时间有关的操作，想要延迟执行，就可以用replayLazily。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 202,056评论 5赞 474
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 84,842评论 2赞 378
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 148,938评论 0赞 335
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,296评论 1赞 272
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,292评论 5赞 363
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,413评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,824评论 3赞 393
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,493评论 0赞 256
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,686评论 1赞 295
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,502评论 2赞 318
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,553评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,281评论 4赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,820评论 3赞 305
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,873评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,109评论 1赞 258
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,699评论 2赞 348
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,257评论 2赞 341

推荐阅读更多精彩内容