ReactiveCocoa 源码解析之核心流程

本文基于ReactiveObjC 3.1.0版本进行解析。共分两部分：

一、核心流程。

二、分析实践。

一、核心流程

创建信号

先来看看创建信号的过程吧：

// 创建信号
RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
        [subscriber sendNext:@"1"];
        [subscriber sendNext:@"2"];
        [subscriber sendNext:@"3"];

        [subscriber sendCompleted];
        return [RACDisposable disposableWithBlock:^{}];
    }];

// 订阅信号
[signal subscribeNext:^(id  _Nullable x) {
    NSLog(@"next: %@", x);
} error:^(NSError * _Nullable error) {
    NSLog(@"error: %@", error);
} completed:^{
    NSLog(@"completed");
}];

这一段代码，相信对于使用过ReactiveCocoa的同学，都不陌生。就是创建了一个信号，这个信号发射了1、2、3以及一个completed。接着看看 createSignal 的过程。

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe];
}

RACSignal 内部创建信号的时候，实际上调用的是 RACDynamicSignal 创建信号的过程。同时我们也可以看到，传递的参数是外面的这个block:

^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
    [subscriber sendNext:@"1"];
    [subscriber sendNext:@"2"];
    [subscriber sendNext:@"3"];
    [subscriber sendCompleted];
    return [RACDisposable disposableWithBlock:^{}];
}

这里只需要有个印象，参数是外部的一个block。

继续看 RACDynamicSignal 创建信号的过程:

@interface RACDynamicSignal ()

// The block to invoke for each subscriber.
@property (nonatomic, copy, readonly) RACDisposable * (^didSubscribe)(id<RACSubscriber> subscriber);

@end

@implementation RACDynamicSignal

#pragma mark Lifecycle

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    RACDynamicSignal *signal = [[self alloc] init];
    signal->_didSubscribe = [didSubscribe copy];
    return [signal setNameWithFormat:@"+createSignal:"];
}

// 其他暂时省略的代码。
@end

通过这部分代码我们可以很清晰地看到， RACDynamicSignal 这个类有一个 block 类型属性 didSubscribe ，并且在 createSignal 的时候将外部传递进来的参数：block对象，赋值给了这个 didSubscribe 属性。

到此为止。创建信号的过程结束。我们看到，这一步的核心动作是 RACDynamicSignal 这个类的对象去持有外部传进来的 block 类型的对象： didSubscribe 。

看完创建信号的过程，接下来就是订阅这个信号了。

订阅信号

订阅信号的过程如下：

[signal subscribeNext:^(id  _Nullable x) {
    NSLog(@"next: %@", x);
} error:^(NSError * _Nullable error) {
    NSLog(@"error: %@", error);
} completed:^{
    NSLog(@"completed");
}];

然后我们再来看看 RACSignal 的 subscribeNext 相关的方法：

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

发现了一个 RACSubscriber ，看名字我们知道，这个才是真正的订阅者。订阅的时候，将这个订阅者作为参数传递给了 signal 的 subscribe 方法。继续看 subscribe 方法实现。

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCAssert(NO, @"This method must be overridden by subclasses");
    return nil;
}

哈哈，RACSignal的 subscribe 方法说，我没有实现，请找找我的子类实现吧。我们回头看看创建信号的时候，其实创建的是 RACDynamicSignal 类的对象。OK，我们看看 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;
}

我们先过滤掉暂时不关心的代码。看到这么一段调用：

RACDisposable *innerDisposable = self.didSubscribe(subscriber);

额，绕来绕去，最终我们发现，订阅的过程实际上是去执行我们创建信号时，传递进去的block类型的参数。并且将 RACSubscriber 类型的对象 subscriber 作为参数传递进去了。并且注意到这个 subscriber 是遵循 RACSubscriber 协议的。

探究到底，我们再来看看 RACSubscriber 协议：

@protocol RACSubscriber <NSObject>
@required

/// 发送value给订阅者.
///
/// value - 可以为nil的value.
- (void)sendNext:(nullable id)value;

/// 发送error给订阅者.
///
/// error - 可以为nil的error.
///
/// 当error发生的时候，会取消所有的订阅动作，订阅者将不再收到任何消息。
- (void)sendError:(nullable NSError *)error;

/// 发送completed给订阅者
///
/// 当completed发生的时候，也会取消所有的订阅。
- (void)sendCompleted;

/// 管理每一次订阅。
/// 告诉订阅者：发生了订阅行为。并将这次订阅行为相关的 `Disposable` 传给订阅者.
/// 以便后续管理取消订阅的流程。
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable;

@end

回过头来看看我们创建信号的过程。我们传递给 RACDynamicSignal 对象的 block 内就有发射 value 、 completed 。

[subscriber sendNext:@"1"];
[subscriber sendNext:@"2"];
[subscriber sendNext:@"3"];
[subscriber sendCompleted];

这里发送了三个 value ：@"1"、@"2"、@"3"，以及一个 completed 。

到这里，订阅者接收了这几个value以及一个completed，订阅的过程基本结束了。还有一点，外部怎么拿到我们的这个value以及completed。留待下一节详细讲解。

到这里，总结一下信号创建及订阅的核心流程：

1、创建RACDynamicSignal信号。

2、并且将一个block类型的对象作为参数，传递给RACDynamicSignal对象的属性didSubscribe。

3、创建一个订阅者RACSubscriber，这个订阅者实现了RACSubscriber协议。

4、执行RACDynamicSignal对象的didSubscribe，并将第3步创建的订阅者作为参数传递给didSubscribe。

5、在didSubscribe这个block内部，传递进来的订阅者RACSubscriber发送value，或者发送error，或者发送completed消息。

6、实现了RACSubscriber协议的订阅者，转而通过自身的block属性，将value、error、completed传递给外部。