ReactCocoa源码分析1
2017-07-21 本文已影响155人
闹鬼的金矿
问题1
在看代码的时候，发现代码中使用到了
- (void)viewDidload
{
[super viewDidload];
[self bindData];
}
- (void)bindData
{
[[RACObserve(self, propertyA) ignore:nil]
subscribeNext:^(NSArray *dataA) {
NSLog(@"use dataA");
}];
}
但是在这个类的propertyA是在init之后去设置的，在viewDidload之前。也就是在使用RAC订阅属性变化信号之前，但是use dataA打印出来了。猜测RACObserve宏生成信号在调用subscribeNext中，直接就调用了dataA的block的逻辑。但是感觉比较奇怪，不应该是propertyA变化的时候才会调用dataA的block的逻辑吗。
现在具体看一下，一个信号的创建和订阅的源码：
信号创建：
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))
didSubscribe {
return [RACDynamicSignal createSignal:didSubscribe];
}
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))
didSubscribe {
RACDynamicSignal *signal = [[self alloc] init];
signal->_didSubscribe = [didSubscribe copy];
return [signal setNameWithFormat:@"+createSignal:"];
}
在创建一个信号的时候，会传进来一个叫didSubscribe的block，该信号会把它存下来。
信号订阅
RACSignal的subscribeNext方法:
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
NSCParameterAssert(nextBlock != NULL);
RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
return [self subscribe:o];
}
在singal的subscribeNext中，生成了一个subscriber。
+ (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;
}
subscriber保存了nextBlock，errorBlock，completedBlock等数据信息
接着看signal的subscribe方法，改方法的参数是subscribeNext方法中生成的subscriber对象
- (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:^{
RACDisposable *innerDisposable = self.didSubscribe(subscriber);
[disposable addDisposable:innerDisposable];
}];
[disposable addDisposable:schedulingDisposable];
}
return disposable;
}
- (RACDisposable *)schedule:(void (^)(void))block {
NSCParameterAssert(block != NULL);
if (RACScheduler.currentScheduler == nil) return [self.backgroundScheduler schedule:block];
block();
return nil;
}
在signal的subscribe方法中，调用了RACScheduler.subscriptionScheduler schedule 方法，直接就将传入的block调用了，最终调用了signal的didSubscribe block，将subscriber传入。
再看一下RACObserve在生成一个signal的时候，传入的didSubscribe block逻辑的怎样的，以下是RACObserve相关源码：
#define RACObserve(TARGET, KEYPATH) \
({ \
_Pragma("clang diagnostic push") \
_Pragma("clang diagnostic ignored \"-Wreceiver-is-weak\"") \
__weak id target_ = (TARGET); \
[target_ rac_valuesForKeyPath:@keypath(TARGET, KEYPATH) observer:self];
\
_Pragma("clang diagnostic pop") \
})
在NSObject的RACPropertySubscribing分类中定义rac_valuesForKeyPath:observer:self:方法
继续：
- (RACSignal *)rac_valuesForKeyPath:(NSString *)keyPath observer:(__weak
NSObject *)observer {
return [[[self
rac_valuesAndChangesForKeyPath:keyPath options:NSKeyValueObservingOptionInitial observer:observer]
map:^(RACTuple *value) {
// -map: because it doesn't require the block trampoline that -reduceEach: uses
return value[0];
}]
setNameWithFormat:@"RACObserve(%@, %@)", self.rac_description, keyPath];
}
继续：
- (RACSignal *)rac_valuesAndChangesForKeyPath:(NSString *)keyPath options:(
NSKeyValueObservingOptions)options observer:(__weak NSObject *)
weakObserver {
NSObject *strongObserver = weakObserver;
keyPath = [keyPath copy];
NSRecursiveLock *objectLock = [[NSRecursiveLock alloc] init];
objectLock.name = @"
org.reactivecocoa.ReactiveCocoa.NSObjectRACPropertySubscribing";
__weak NSObject *weakSelf = self;
RACSignal *deallocSignal = [[RACSignal
zip:@[
self.rac_willDeallocSignal,
strongObserver.rac_willDeallocSignal ?: [RACSignal never]
]]
doCompleted:^{
// Forces deallocation to wait if the object variables are currently
// being read on another thread.
[objectLock lock];
@onExit {
[objectLock unlock];
};
}];
//重点关注这里，createSignal之后的参数就是该信号的didSubscribe block逻辑了。
return [[[RACSignal
createSignal:^ RACDisposable * (id<RACSubscriber> subscriber) {
// Hold onto the lock the whole time we're setting up the KVO
// observation, because any resurrection that might be caused by our
// retaining below must be balanced out by the time -dealloc returns
// (if another thread is waiting on the lock above).
[objectLock lock];
@onExit {
[objectLock unlock];
};
__strong NSObject *observer __attribute__((objc_precise_lifetime))
= weakObserver;
__strong NSObject *self __attribute__((objc_precise_lifetime)) =
weakSelf;
if (self == nil) {
[subscriber sendCompleted];
return nil;
}
return [self rac_observeKeyPath:keyPath options:options observer:
observer block:^(id value, NSDictionary *change, BOOL
causedByDealloc, BOOL affectedOnlyLastComponent) {
[subscriber sendNext:RACTuplePack(value, change)];
}];
}]
takeUntil:deallocSignal]
setNameWithFormat:@"%@ -rac_valueAndChangesForKeyPath: %@ options: %lu
observer: %@", self.rac_description, keyPath, (unsigned long)options,
strongObserver.rac_description];
}
可以看到在RACObserver宏定义的signal的didSubscriber block中又调用了rac_observeKeyPath:keyPath options: observer: block
继续(太长了只贴重点)
- (RACDisposable *)rac_observeKeyPath:(NSString *)keyPath options:(
NSKeyValueObservingOptions)options observer:(__weak NSObject *)
weakObserver block:(void (^)(id, NSDictionary *, BOOL, BOOL))block {
NSCParameterAssert(block != nil);
NSCParameterAssert(keyPath.rac_keyPathComponents.count > 0);
//省略数十行
// Call the block with the initial value if needed.
if ((options & NSKeyValueObservingOptionInitial) != 0) {
id initialValue = [self valueForKeyPath:keyPath];
NSDictionary *initialChange = @{
NSKeyValueChangeKindKey: @(NSKeyValueChangeSetting),
NSKeyValueChangeNewKey: initialValue ?: NSNull.null,
};
block(initialValue, initialChange, NO, keyPathHasOneComponent);
}
//省略数十行
}
说明一下，options是NSKeyValueObservingOptions属于NS_OPTIONS
typedef NS_OPTIONS(NSUInteger, NSKeyValueObservingOptions) {
NSKeyValueObservingOptionNew = 0x01,
NSKeyValueObservingOptionOld = 0x02,
NSKeyValueObservingOptionInitial NS_ENUM_AVAILABLE(10_5, 2_0) = 0x04,
NSKeyValueObservingOptionPrior NS_ENUM_AVAILABLE(10_5, 2_0) = 0x08
};
在以上方法中，它判断了，传入的options是否是NSKeyValueObservingOptionInitial类型，而在调用rac_observeKeyPath: options: observer: block:的时候，option就是传的NSKeyValueObservingOptionInitial，所以会直接调用传进来的block，在rac_valuesAndChangesForKeyPath: options: observer:中调用rac_observeKeyPath: options: observer: block:的时候传入block里面的逻辑是这样：
[subscriber sendNext:RACTuplePack(value, change)];
综上所述，RACObserver生成的signal在调用subscribeNext方法订阅该信号的时候，会直接调用一次订阅信号之后next block的逻辑，所以即便是属性变化之后订阅属性变化信号，它也会默认先调用一次next block的逻辑。
正常kvo检测转化成信号的逻辑：
在RACObserver初始化的过程中，
- (RACDisposable *)rac_observeKeyPath:(NSString *)keyPath options:(
NSKeyValueObservingOptions)options observer:(__weak NSObject *)
weakObserver block:(void (^)(id, NSDictionary *, BOOL, BOOL))block {
NSCParameterAssert(block != nil);
NSCParameterAssert(keyPath.rac_keyPathComponents.count > 0);
//省略数十行
NSKeyValueObservingOptions trampolineOptions = (options |
NSKeyValueObservingOptionPrior) & ~NSKeyValueObservingOptionInitial;
RACKVOTrampoline *trampoline = [[RACKVOTrampoline alloc] initWithTarget:
self observer:strongObserver keyPath:keyPathHead options:trampolineOptions
block:^(id trampolineTarget, id trampolineObserver, NSDictionary *change) {
// If this is a prior notification, clean up all the callbacks added to the
// previous value and call the callback block. Everything else is deferred
// until after we get the notification after the change.
if ([change[NSKeyValueChangeNotificationIsPriorKey] boolValue]) {
[firstComponentDisposable() dispose];
if ((options & NSKeyValueObservingOptionPrior) != 0) {
block([trampolineTarget valueForKeyPath:keyPath], change, NO,
keyPathHasOneComponent);
}
return;
}
// From here the notification is not prior.
NSObject *value = [trampolineTarget valueForKey:keyPathHead];
// If the value has changed but is nil, there is no need to add callbacks to
// it, just call the callback block.
if (value == nil) {
block(nil, change, NO, keyPathHasOneComponent);
return;
}
// From here the notification is not prior and the value is not nil.
// Create a new firstComponentDisposable while getting rid of the old one at
// the same time, in case this is being called concurrently.
RACDisposable *oldFirstComponentDisposable = [
firstComponentSerialDisposable swapInDisposable:[RACCompoundDisposable
compoundDisposable]];
[oldFirstComponentDisposable dispose];
addDeallocObserverToPropertyValue(value);
// If there are no further key path components, there is no need to add the
// other callbacks, just call the callback block with the value itself.
if (keyPathHasOneComponent) {
block(value, change, NO, keyPathHasOneComponent);
return;
}
// The value has changed, is not nil, and there are more key path components
// to consider. Add the callbacks to the value for the remaining key path
// components and call the callback block with the current value of the full
// key path.
addObserverToValue(value);
block([value valueForKeyPath:keyPathTail], change, NO, keyPathHasOneComponent);
}];
// Stop the KVO observation when this one is disposed of.
[disposable addDisposable:trampoline];
//省略数十行
}
在该方法中生成了一个RACKVOTrampoline中间对象，看它的源码
- (id)initWithTarget:(__weak NSObject *)target observer:(__weak NSObject *)
observer keyPath:(NSString *)keyPath options:(NSKeyValueObservingOptions)
options block:(RACKVOBlock)block {
NSCParameterAssert(keyPath != nil);
NSCParameterAssert(block != nil);
NSObject *strongTarget = target;
if (strongTarget == nil) return nil;
self = [super init];
if (self == nil) return nil;
_keyPath = [keyPath copy];
_block = [block copy];
_weakTarget = target;
_unsafeTarget = strongTarget;
_observer = observer;
[RACKVOProxy.sharedProxy addObserver:self forContext:(__bridge void *)self];
[strongTarget addObserver:RACKVOProxy.sharedProxy forKeyPath:self.keyPath
options:options context:(__bridge void *)self];
[strongTarget.rac_deallocDisposable addDisposable:self];
[self.observer.rac_deallocDisposable addDisposable:self];
return self;
}
- (void)dealloc {
[self dispose];
}
#pragma mark Observation
- (void)dispose {
NSObject *target;
NSObject *observer;
@synchronized (self) {
_block = nil;
// The target should still exist at this point, because we still need to
// tear down its KVO observation. Therefore, we can use the unsafe
// reference (and need to, because the weak one will have been zeroed by
// now).
target = self.unsafeTarget;
observer = self.observer;
_unsafeTarget = nil;
_observer = nil;
}
[target.rac_deallocDisposable removeDisposable:self];
[observer.rac_deallocDisposable removeDisposable:self];
[target removeObserver:RACKVOProxy.sharedProxy forKeyPath:self.keyPath
context:(__bridge void *)self];
[RACKVOProxy.sharedProxy removeObserver:self forContext:(__bridge void *)self];
}
//系统的代理方法，其实是由RACKVOProxy.sharedProxy对象转发的，RACKVOProxy.sharedProxy才是真正处理系统消息的对象。
- (void)observeValueForKeyPath:(NSString *)keyPath ofObject:(id)object change:(
NSDictionary *)change context:(void *)context {
if (context != (__bridge void *)self) {
[super observeValueForKeyPath:keyPath ofObject:object change:change
context:context];
return;
}
RACKVOBlock block;
id observer;
id target;
@synchronized (self) {
block = self.block;
observer = self.observer;
target = self.weakTarget;
}
if (block == nil || target == nil) return;
block(target, observer, change);
}
可以看到RACKVOTrampoline对象替代原来使用KVO的对象，作为系统的代理，实现了代理方法。实际上，真正调用系统KVO注册的方法的时候，是往一个叫RACKVOProxy.sharedProxy的全局单例对象注册的。RACKVOTrampoline是具体处理KVO消息的对象，在RACKVOPorxy.shareProxy对象中注册了所有使用RAC KVO的系统消息，再由它转发给具体的RACKVOTrampoline进行处理，而在RACKVOTrampoline处理的时候，调用了RACKVOtrampoline初始化的时候传进来的block。之后在RACKVOTrampoline参数block调用过程中就会调用sendNext方法了，往外面发信号数据。
以下是RACKVOProxy.sharedProxy
@interface RACKVOProxy()
@property (strong, nonatomic, readonly) NSMapTable *trampolines;
@property (strong, nonatomic, readonly) dispatch_queue_t queue;
@end
@implementation RACKVOProxy
+ (instancetype)sharedProxy {
static RACKVOProxy *proxy;
static dispatch_once_t onceToken;
dispatch_once(&onceToken, ^{
proxy = [[self alloc] init];
});
return proxy;
}
- (instancetype)init {
self = [super init];
if (self == nil) return nil;
_queue = dispatch_queue_create("org.reactivecocoa.ReactiveCocoa.RACKVOProxy
", DISPATCH_QUEUE_SERIAL);
_trampolines = [NSMapTable strongToWeakObjectsMapTable];
return self;
}
- (void)addObserver:(__weak NSObject *)observer forContext:(void *)context {
NSValue *valueContext = [NSValue valueWithPointer:context];
dispatch_sync(self.queue, ^{
[self.trampolines setObject:observer forKey:valueContext];
});
}
- (void)removeObserver:(NSObject *)observer forContext:(void *)context {
NSValue *valueContext = [NSValue valueWithPointer:context];
dispatch_sync(self.queue, ^{
[self.trampolines removeObjectForKey:valueContext];
});
}
- (void)observeValueForKeyPath:(NSString *)keyPath ofObject:(id)object change:(
NSDictionary *)change context:(void *)context {
NSValue *valueContext = [NSValue valueWithPointer:context];
__block NSObject *trueObserver;
dispatch_sync(self.queue, ^{
trueObserver = [self.trampolines objectForKey:valueContext];
});
if (trueObserver != nil) {
[trueObserver observeValueForKeyPath:keyPath ofObject:object change:
change context:context];
}
}
RACKVOProxy.sharedProxy管理了整个RAC 中KVO的处理系统KVO消息的中间对象和系统KVO消息的转发。
综合上面的代码可以看出，正是由于各种中间对象替用户实现了代理方法起了代理对象的作用，用户才能把代码写的更加紧凑清晰。
问题2
看以下代码，假设combineLatest之后得到的信号是A
[[RACSignal combineLatest:@[[RACObserve(self, propertyA) ignore:nil], [
RACObserve(self, propertyB) ignore:nil]]] subscribeNext:^(RACTuple *tuple) {
}];
1.使用combineLatest的时候，第一次订阅会不会触发subscribeNext后面的block
2.combineLatest中的信号，是同时调用了sendNext之后会触发A调用sendNext，还是只需要其中有一个信号调用了sendNext会触发A调用sendNext
看一下combineLatest源码：
+ (RACSignal *)combineLatest:(id<NSFastEnumeration>)signals {
return [[self join:signals block:^(RACSignal *left, RACSignal *right) {
return [left combineLatestWith:right];
}] setNameWithFormat:@"+combineLatest: %@", signals];
}
继续 join: block:
+ (instancetype)join:(id<NSFastEnumeration>)streams block:(RACStream * (^)(id,
id))block {
//第一段
RACStream *current = nil;
// Creates streams of successively larger tuples by combining the input
// streams one-by-one.
for (RACStream *stream in streams) {
// For the first stream, just wrap its values in a RACTuple. That way,
// if only one stream is given, the result is still a stream of tuples.
if (current == nil) {
current = [stream map:^(id x) {
return RACTuplePack(x);
}];
continue;
}
current = block(current, stream);
}
if (current == nil) return [self empty];
//第二段
return [current map:^(RACTuple *xs) {
// Right now, each value is contained in its own tuple, sorta like:
//
// (((1), 2), 3)
//
// We need to unwrap all the layers and create a tuple out of the result.
NSMutableArray *values = [[NSMutableArray alloc] init];
while (xs != nil) {
[values insertObject:xs.last ?: RACTupleNil.tupleNil atIndex:0];
xs = (xs.count > 1 ? xs.first : nil);
}
return [RACTuple tupleWithObjectsFromArray:values];
}];
}
这部分代码分2段，第一段是将两个信号合并的逻辑，具体的合并逻辑是由外面传进来的block确定的。第二段是通过map将信号的值重新做了处理，第一段得到的信号属于signal of signals的类型，第二段将它打平。
再看一下combineLatestWith:方法
- (RACSignal *)combineLatestWith:(RACSignal *)signal {
NSCParameterAssert(signal != nil);
return [[RACSignal createSignal:^(id<RACSubscriber> subscriber) {
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
__block id lastSelfValue = nil;
__block BOOL selfCompleted = NO;
__block id lastOtherValue = nil;
__block BOOL otherCompleted = NO;
void (^sendNext)(void) = ^{
@synchronized (disposable) {
if (lastSelfValue == nil || lastOtherValue == nil) return;
[subscriber sendNext:RACTuplePack(lastSelfValue, lastOtherValue)];
}
};
RACDisposable *selfDisposable = [self subscribeNext:^(id x) {
@synchronized (disposable) {
lastSelfValue = x ?: RACTupleNil.tupleNil;
sendNext();
}
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
@synchronized (disposable) {
selfCompleted = YES;
if (otherCompleted) [subscriber sendCompleted];
}
}];
[disposable addDisposable:selfDisposable];
RACDisposable *otherDisposable = [signal subscribeNext:^(id x) {
@synchronized (disposable) {
lastOtherValue = x ?: RACTupleNil.tupleNil;
sendNext();
}
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
@synchronized (disposable) {
otherCompleted = YES;
if (selfCompleted) [subscriber sendCompleted];
}
}];
[disposable addDisposable:otherDisposable];
return disposable;
}] setNameWithFormat:@"[%@] -combineLatestWith: %@", self.name, signal];
}
在以上代码中，调用了当前信号的subscribeNext方法，同时也调用了需要合并的信号的subscribeNext方法。subscribeNext方法block中调用了sendNext block，这个block是在combineLatestWith中定义，判断两个信号是否已经调用过sendNext，如果都同时掉用过sendNext就会触发combineLatest信号调用didSubscribe block，最终触发订阅combineLatest信号的传入的subscribeNext后的block。
综合上面的分析，类似于以下的使用方式
[[RACSignal combineLatest:@[[RACObserve(self, propertyA) ignore:nil], [RACObserve(self, propertyB) ignore:nil]]] subscribeNext:^(RACTuple *tuple) {
}];
第一次订阅就会触发subscribeNext之后的block逻辑，并且是RACObserve这种类型的combineLatest才会，最上面已经分析了RACObserver生成的信号在第一次订阅调用的时候信号就会调用sendNext。
