日常使用

RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber){
    /// STEP 1
    [subscriber sendNext:@(1)];
    [subscriber sendNext:@(2)];
    [subscriber sendNext:@(3)];
    [subscriber sendCompleted];
    return [RACDisposable disposableWithBlock:^{
        /// STEP 5
        NSLog(@"signal dispose");
    }];
}];

RACDisposable *disposable = [signal subscribeNext:^(id x) {
    /// STEP 2
    NSLog(@"received value = %@", x);
} error:^(NSError *error) {
    /// STEP 3
    NSLog(@"received error: %@", error);
} completed:^{
    /// STEP 4
    NSLog(@"received completed");
}];

[disposable dispose];

RACSignal 从创建到给订阅发送事件步骤归纳

调用RACSignal createSignal 的方法，返回子类 RACDynamicSignal 对象并保存闭包 didSubscribe

@implementation RACSignal
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe]; // 返回 RACDynamicSignal 类型的对象
}

@interface RACDynamicSignal : RACSignal
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe;
@end

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

订阅信号，调用RACSignal subscribeNext等方法，在该方法中会创建 RACSubscriber 订阅者对象

- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    return [self subscribe:o];
}

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

步骤1创建的 RACDynamicSignal 调用 subscribe 方法，创建 RACCompoundDisposable 和 RACPassthroughSubscriber 对象，RACPassthroughSubscriber 会保存步骤1返回的 RACDynamicSignal（signal属性），步骤2中的 RACSubscriber（innerSubscriber属性）

@implementation RACDynamicSignal

- (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:^{
            //触发block，执行后续的信号发送操作，并注册丢弃信号的响应事件
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];

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

执行 RACDynamicSignal 保存的 didSubscribe 闭包，闭包内调用 RACPassthroughSubscriber sendNext，sendError，sendCompleted 方法。

@implementation RACPassthroughSubscriber

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

- (void)sendError:(NSError *)error {
    if (self.disposable.disposed) return;

    if (RACSIGNAL_ERROR_ENABLED()) {
        RACSIGNAL_ERROR(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description), cleanedDTraceString(error.description));
    }

    [self.innerSubscriber sendError:error];
}

- (void)sendCompleted {
    if (self.disposable.disposed) return;

    if (RACSIGNAL_COMPLETED_ENABLED()) {
        RACSIGNAL_COMPLETED(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description));
    }

    [self.innerSubscriber sendCompleted];
}

RACPassthroughSubscriber 执行保存的 innerSubscriber 对应的 sendNext，sendError，sendCompleted 方法

@implementation RACSubscriber

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

        nextBlock(value);
    }
}

- (void)sendError:(NSError *)e {
    @synchronized (self) {
        void (^errorBlock)(NSError *) = [self.error copy];
        [self.disposable dispose];

        if (errorBlock == nil) return;
        errorBlock(e);
    }
}

- (void)sendCompleted {
    @synchronized (self) {
        void (^completedBlock)(void) = [self.completed copy];
        [self.disposable dispose];

        if (completedBlock == nil) return;
        completedBlock();
    }
}