日常使用

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 從創(chuàng)建到給訂閱發(fā)送事件步驟歸納

1. 調(diào)用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:"];
}

2. 訂閱信號，調(diào)用RACSignal subscribeNext等方法，在該方法中會創(chuàng)建 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;
}

3. 步驟1創(chuàng)建的 RACDynamicSignal 調(diào)用 subscribe 方法，創(chuàng)建 RACCompoundDisposable 和 RACPassthroughSubscriber 對象，RACPassthroughSubscriber 會保存 步驟1返回的 RACDynamicSignal（signal屬性），步驟2中的 RACSubscriber（innerSubscriber屬性）

@implementation RACDynamicSignal

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCParameterAssert(subscriber != nil);
    //處理信號丟棄
    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    //處理信號發(fā)送 
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

    if (self.didSubscribe != NULL) {
        RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            //觸發(fā)block，執(zhí)行后續(xù)的信號發(fā)送操作，并注冊丟棄信號的響應(yīng)事件
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];

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

4. 執(zhí)行 RACDynamicSignal 保存的 didSubscribe 閉包，閉包內(nèi)調(diào)用 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];
}

5. RACPassthroughSubscriber 執(zhí)行保存的 innerSubscriber 對應(yīng)的 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();
    }
}