RACDisposableRACSubscriber

Posted 2021-10-07 王飞飞不会飞

在了解RAC的过程中，RACSignal的工作处理流程中，少不了RACDisposable和RACSubscriber的参与。本文通过前文的使用代码、解读源码，来探索RACDisposable和RACSubscriber的工作原理。

RACDisposable

• 翻译

disposable
英 [dɪˈspəʊzəbl]   美 [dɪˈspoʊzəbl]  
adj.
可任意处理的;一次性的;用后即丢弃的;可动用的;可自由支配的
n.
〈美口〉使用后随即抛掉的东西（尤指容器等）

• 使用

- (RACSignal *)signalTest {
    RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        NSLog(@"this is create block");
        [subscriber sendNext:@"发送Next"];
        RACDisposable *testDisposable = [RACDisposable disposableWithBlock:^{
            NSLog(@"this is dispose creat block");
        }];
        return testDisposable;
    }];
    
    RACDisposable *disposable = [signal subscribeNext:^(id x) {
        NSLog(@"this is subscribe block %@",x);
    }];
    
    [disposable dispose];

    return signal;
}

这段代码中出现了2个disposable，RACSignal的didSubscribe 中返回的disposable，RACSignal的subscribeNext方法的返回的disposable。这两个一个是往RACSignal内部传入一个disposable，一个是内部返回一个disposable，很容易给人一种是同一个disposable的错觉，经过打印地址，可以发现并不是同一个disposable，那其中有什么联系么？或者说这内部做了什么操作呢？

• 源码

先看看RACDisposable的初始化方法吧

- (id)initWithBlock:(void (^)(void))block {
	NSCParameterAssert(block != nil);

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

	_disposeBlock = (void *)CFBridgingRetain([block copy]); 
	OSMemoryBarrier();

	return self;
}

初始化方法很简单，就是吧block存起来。那这个block在什么时候调用呢？

1.RACDisposable的dispose方法

2.RACSubscriber的dealloc方法

//1.在 RACDisposable中
- (void)dispose {
	void (^disposeBlock)(void) = NULL;

	while (YES) {
		void *blockPtr = _disposeBlock;
		if (OSAtomicCompareAndSwapPtrBarrier(blockPtr, NULL, &_disposeBlock)) {
			if (blockPtr != (__bridge void *)self) {
				disposeBlock = CFBridgingRelease(blockPtr);
			}

			break;
		}
	}

	if (disposeBlock != nil) disposeBlock();
}

//2.在 RACSubscriber中
- (void)dealloc {
	[self.disposable dispose];
}

接下来再看看RACSignal的subscribeNext方法中返回的RACDisposable的来源。

// subscribeNext方法里面调用subscribe方法，所以直接来到subscribe这个方法中来
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
	NSCParameterAssert(subscriber != nil);
    //新建一个RACCompoundDisposable作为返回值。
	RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
	subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

	if (self.didSubscribe != NULL) {
		RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            //didSubscribe Block返回来的disposable
			RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            //添加到要返回的disposable中
			[disposable addDisposable:innerDisposable];
		}];

		[disposable addDisposable:schedulingDisposable];
	}
	//直接返回
	return disposable;
}

这里看到返回的disposable和didSubscribe Block返回的innerDisposable并不是一个。也验证了使用代码中两个RACDisposable不是同一个了。

RACSubscriber

• 使用

同RACDisposable的使用，在RACSignal的didSubscribe Block中作为回掉参数，可以使用send Next方法。

• 源码

didSubscribe中的subscriber类型为RACPassthroughSubscriber，所以来到RACPassthroughSubscriber的 sendNext方法中

- (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)sendNext:(id)value {
	@synchronized (self) {
		void (^nextBlock)(id) = [self.next copy];
		if (nextBlock == nil) return;

		nextBlock(value);
	}
}

这里执行了nextBlock。

我们看看初始化方法

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

我们在来看看RACPassthroughSubscriber的初始化方法

- (instancetype)initWithSubscriber:(id<RACSubscriber>)subscriber signal:(RACSignal *)signal disposable:(RACCompoundDisposable *)disposable {
	NSCParameterAssert(subscriber != nil);

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

	_innerSubscriber = subscriber;
	_signal = signal;
	_disposable = disposable;

	[self.innerSubscriber didSubscribeWithDisposable:self.disposable];
	return self;
}

- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)otherDisposable {
	if (otherDisposable.disposed) return;

	RACCompoundDisposable *selfDisposable = self.disposable;
	[selfDisposable addDisposable:otherDisposable];

	@unsafeify(otherDisposable);

	// If this subscription terminates, purge its disposable to avoid unbounded
	// memory growth.
	[otherDisposable addDisposable:[RACDisposable disposableWithBlock:^{
		@strongify(otherDisposable);
		[selfDisposable removeDisposable:otherDisposable];
	}]];
}

在往里看

- (void)addDisposable:(RACDisposable *)disposable {
	NSCParameterAssert(disposable != self);
	if (disposable == nil || disposable.disposed) return;

	BOOL shouldDispose = NO;

	OSSpinLockLock(&_spinLock);
	{
		if (_disposed) {
			shouldDispose = YES;
		} else {
			#if RACCompoundDisposableInlineCount
			for (unsigned i = 0; i < RACCompoundDisposableInlineCount; i++) {
				if (_inlineDisposables[i] == nil) {
					_inlineDisposables[i] = disposable;
					goto foundSlot;
				}
			}
			#endif

			if (_disposables == NULL) _disposables = RACCreateDisposablesArray();
			CFArrayAppendValue(_disposables, (__bridge void *)disposable);

			if (RACCOMPOUNDDISPOSABLE_ADDED_ENABLED()) {
				RACCOMPOUNDDISPOSABLE_ADDED(self.description.UTF8String, disposable.description.UTF8String, CFArrayGetCount(_disposables) + RACCompoundDisposableInlineCount);
			}

		#if RACCompoundDisposableInlineCount
		foundSlot:;
		#endif
		}
	}
	OSSpinLockUnlock(&_spinLock);

	// Performed outside of the lock in case the compound disposable is used
	// recursively.
	if (shouldDispose) [disposable dispose];
}

这里注意这里也有dispose方法，所以Disposable的block在这里应该也是有可能触发的。

• 图解

根据以上的源码，这几个类的关系已经浮出水面了。这里逻辑比较绕，多看几遍源码才能理解。

这里有几点需要注意

1.RACPassthroughSubscriber不是RACSubscriber的子类，两者同遵循RACSubscriber协议 

2.RACSubscriber的didSubscribeWithDisposable方法，暂时理解为捕获disposable。其中直接调用selfDisposable 的addDisposable方法。selfDisposable也是RACCompoundDisposable类型的，RACCompoundDisposable中有_inlineDisposables数组，这里应该是吧传入的disposable存起来了。