• RACSubject :信号提供者,自己可以充当信号,又能够发送信号

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

/// Returns a new subject.
+ (instancetype)subject;

// Redeclaration of the RACSubscriber method. Made in order to specify a generic type.
- (void)sendNext:(nullable ValueType)value;

@end

基本使用

#import "ViewController.h"
#import <ReactiveObjC/ReactiveObjC.h>

@interface ViewController ()

@end

@implementation ViewController


- (void)touchesBegan:(NSSet<UITouch *> *)touches withEvent:(UIEvent *)event {
    
    // 1.订阅信号
    RACSubject *subject = [RACSubject subject];
    
    // 2.订阅信号
    // 不同的信号，订阅方式不一样（因为类型不一样，所以调用方法不一样）
    [subject subscribeNext:^(id  _Nullable x) {
       
        NSLog(@"第1个订阅信号中的内容是: %@", x);
    }];
    
    [subject subscribeNext:^(id  _Nullable x) {
        
        NSLog(@"第2个订阅信号中的内容是: %@", x);
    }];
    
    [subject subscribeNext:^(id  _Nullable x) {
        
        NSLog(@"第3个订阅信号中的内容是: %@", x);
    }];
    
    // 3.发送信号
    // 遍历出所有的订阅者,调用nextBlock
    [subject sendNext:@"subject的信号"];
}

过程分析

RACSubject.png

• 创建信号

// RACSubject
// 创建一个容量为1的数组
+ (instancetype)subject {
    return [[self alloc] init];
}

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

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

• 订阅信号

// RACSignal
// 订阅信号 调用父类的方法 RACSignal 
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    
    // 创建订阅者    在内部 保存了 nextBlock
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    
    return [self subscribe:o]; // 注意此时的 self 是 RACSubject
}

// RACSubject
// 保存订阅者

// This should only be used while synchronized on `self`.
@property (nonatomic, strong, readonly) NSMutableArray *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;
}

• 发送信号

// RACSubject 

// This should only be used while synchronized on `self`.
@property (nonatomic, strong, readonly) NSMutableArray *subscribers;

// 遍历每个一个订阅都 进行逐一发送消息
- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {
    NSArray *subscribers;
    @synchronized (self.subscribers) {
        subscribers = [self.subscribers copy];
    }

    for (id<RACSubscriber> subscriber in subscribers) {
        block(subscriber);
    }
}

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

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

        nextBlock(value);
    }
}