ReactiveCocoa 源码阅读记录。

 1     RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
 2         [subscriber sendNext:@"1"];
 3         [subscriber sendCompleted];
 4         return nil;
 5     }];
 6     
 7     [signal subscribeNext:^(id x) {
 8         
 9     } error:^(NSError * _Nullable error) {
10         
11     } completed:^{
12         
13     }];
14 /*************    对应的源码    **************/
15 // 生成RACDynamicSignal。即生成RACSignal的子类
16 + (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
17     return [RACDynamicSignal createSignal:didSubscribe];
18 }
19 
20 // 在子类中保存 didSubscribe block,每一次订阅都会执行
21 + (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
22     RACDynamicSignal *signal = [[self alloc] init];
23     signal->_didSubscribe = [didSubscribe copy];
24     return [signal setNameWithFormat:@"+createSignal:"];
25 }
26 
27 /// 在订阅时执行保存在RACDynamicSignal中的didSubscribe block
28 - (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
29     NSCParameterAssert(subscriber != nil);
30 
31     RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
32     subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
33 
34     if (self.didSubscribe != NULL) {
35         RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
36             RACDisposable *innerDisposable = self.didSubscribe(subscriber);
37             [disposable addDisposable:innerDisposable];
38         }];
39 
40         [disposable addDisposable:schedulingDisposable];
41     }
42     
43     return disposable;
44 }
45 
46 // 生成订阅者。并调用RACDynamicSignal中的订阅方法，执行didSubscribe block
47 - (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
48     NSCParameterAssert(nextBlock != NULL);
49     
50     RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
51     return [self subscribe:o];
52 }
53 
54 // 生成的订阅者中保存 nextBlock， errorBlock， completedBlock.
55 + (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
56     RACSubscriber *subscriber = [[self alloc] init];
57 
58     subscriber->_next = [next copy];
59     subscriber->_error = [error copy];
60     subscriber->_completed = [completed copy];
61 
62     return subscriber;
63 }
64 
65 // 当订阅者收到sendNext方法时，就会执行nextBlock
66 - (void)sendNext:(id)value {
67     @synchronized (self) {
68         void (^nextBlock)(id) = [self.next copy];
69         if (nextBlock == nil) return;
70 
71         nextBlock(value);
72     }
73 }

1    /// 事例    
2     RACSubject *sub = [[RACSubject alloc] init];
3     [sub subscribeNext:^(id  _Nullable x) {
4         NSLog(@"%@",a);
5     }];
6     
7     [sub sendNext:@"a"];

 1 //调用时和RACSignal有区别的方法
 2 // 使用 subscribers 保存 订阅者。（RACDynamicSignal 还要执行didSubscrbe Block）
 3 - (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
 4     NSCParameterAssert(subscriber != nil);
 5 
 6     RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
 7     subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
 8 
 9     NSMutableArray *subscribers = self.subscribers;
10     @synchronized (subscribers) {
11         [subscribers addObject:subscriber];
12     }
13     
14     [disposable addDisposable:[RACDisposable disposableWithBlock:^{
15         @synchronized (subscribers) {
16             // Since newer subscribers are generally shorter-lived,search
17             // starting from the end of the list.
18             NSUInteger index = [subscribers indexOfObjectWithOptions:NSEnumerationReverse passingTest:^ BOOL (id<RACSubscriber> obj,NSUInteger index,BOOL *stop) {
19                 return obj == subscriber;
20             }];
21 
22             if (index != NSNotFound) [subscribers removeObjectAtIndex:index];
23         }
24     }]];
25 
26     return disposable;
27 }
28 
29 // 找到需要的订阅者
30 - (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {
31     NSArray *subscribers;
32     @synchronized (self.subscribers) {
33         subscribers = [self.subscribers copy];
34     }
35 
36     for (id<RACSubscriber> subscriber in subscribers) {
37         block(subscriber);
38     }
39 }
40 
41 // 这行nextBlock。
42 - (void)sendNext:(id)value {
43     [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
44         [subscriber sendNext:value];
45     }];
46 }