filter: 过滤
过滤原始信号,如果满足过滤条件转发这个信号,否则忽略这个信号.
示例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@"zhao"];
[subscriber sendNext:@"wang"];
[subscriber sendNext:@"qian"];
[subscriber sendNext:@"wei"];
return [[RACDisposable alloc] init];
}]
filter:^BOOL(id _Nullable value) {
if ([value hasPrefix:@"w"]) {
return YES;
} else {
return NO;
}
}];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"x : %@", x);
}];
输出:
2017-09-29 16:27:41.273049+0800 RAC[7343:336413] x : wang
2017-09-29 16:27:41.273326+0800 RAC[7343:336413] x : wei
实现:
- (__kindof RACStream *)filter:(BOOL (^)(id value))block {
NSCParameterAssert(block != nil);
Class class = self.class;
return [[self flattenMap:^ id (id value) {
if (block(value)) {
return [class return:value];
} else {
return class.empty;
}
}] setNameWithFormat:@"[%@] -filter:", self.name];
}
如果满足过滤条件返回一个调用-[RACReturnSignal return:] 直接同步发送信号,否则返回一个RACEmptySignal
RACReturnSignal
+ (RACSignal *)return:(id)value {
RACReturnSignal *signal = [[self alloc] init];
signal->_value = value;
#ifdef DEBUG
[signal setNameWithFormat:@"+return: %@", value];
#endif
return signal;
}
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
NSCParameterAssert(subscriber != nil);
return [RACScheduler.subscriptionScheduler schedule:^{
[subscriber sendNext:self.value];
[subscriber sendCompleted];
}];
}
上面是一个return信号的实现,在创建过程中保存了vulue,当此信号被订阅时,直接发送value并调用完成信号.
RACEmptySignal
+ (RACSignal *)empty {
#ifdef DEBUG
// Create multiple instances of this class in DEBUG so users can set custom
// names on each.
return [[[self alloc] init] setNameWithFormat:@"+empty"];
#else
static id singleton;
static dispatch_once_t pred;
dispatch_once(&pred, ^{
singleton = [[self alloc] init];
});
return singleton;
#endif
}
#pragma mark Subscription
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
NSCParameterAssert(subscriber != nil);
return [RACScheduler.subscriptionScheduler schedule:^{
[subscriber sendCompleted];
}];
}
上面就是一个空信号的实现,在这个信号被调用时什么也不做直接调用完成信号, 注意在这里区分了release版本和DEBUG版本, 在release版本使用一个单例实现RACEmptySignal.
所以通过RACReturnSignal RACEmptySignal 两个信号对过滤进行转发和忽略.
ignore: 忽略
忽略指定的值
示例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@"wang"];
[subscriber sendNext:@"li"];
[subscriber sendNext:@"fang"];
[subscriber sendNext:@"wang"];
return [[RACDisposable alloc] init];
}] ignore:@"wang"];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"x : %@", x);
}];
输出:
2017-09-29 17:11:37.726000+0800 RAC[7881:371770] x : li
2017-09-29 17:11:37.726229+0800 RAC[7881:371770] x : fang
实现:- (__kindof RACStream *)ignore:(id)value {
return [[self filter:^ BOOL (id innerValue) {
return innerValue != value && ![innerValue isEqual:value];
}] setNameWithFormat:@"[%@] -ignore: %@", self.name, RACDescription(value)];
}
内部是对filter方法的封装, 内部是使用指针地址和 - isEqual:方法判断两个值是否相等, 如果两个值相等则忽略调这个值.
reduceEach:
block参数的个数是动态的,根据元组中的元素个数变化,block每个参数和元组中的每个元素一一对应. block的返回值是根据元组中的元素映射的一个值,其中的逻辑可以根据需求而定.
示例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:RACTuplePack(@1, @2)];
[subscriber sendNext:RACTuplePack(@3, @4)];
return [[RACDisposable alloc] init];
}] reduceEach:^id _Nullable (id value1, id value2){
return @([value1 integerValue] + [value2 integerValue]);
}];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"%ld", [x integerValue]);
}];
输出:
2017-10-03 19:24:33.999636+0800 RAC[28155:1726872] 3
2017-10-03 19:24:33.999854+0800 RAC[28155:1726872] 7
源码:
- (__kindof RACStream *)reduceEach:(id (^)())reduceBlock {
NSCParameterAssert(reduceBlock != nil);
__weak RACStream *stream __attribute__((unused)) = self;
return [[self map:^(RACTuple *t) {
NSCAssert([t isKindOfClass:RACTuple.class], @"Value from stream %@ is not a tuple: %@", stream, t);
return [RACBlockTrampoline invokeBlock:reduceBlock withArguments:t];
}] setNameWithFormat:@"[%@] -reduceEach:", self.name];
}
在源码中有两个断言,一个断言是reduceBlock不能为空,另一个是信号的值必须为RACTuple类型.
reduceEach:内部是对map方法的封装,根据mapBlock的入参返回[RACBlockTrampoline invokeBlock:reduceBlock withArguments:t].
RACBlockTrampoline根据入参RACTuple的count调用block,返回调用block的返回值.
+ (id)invokeBlock:(id)block withArguments:(RACTuple *)arguments {
NSCParameterAssert(block != NULL);
//保存block
RACBlockTrampoline *trampoline = [[self alloc] initWithBlock:block];
return [trampoline invokeWithArguments:arguments];
}
- (id)invokeWithArguments:(RACTuple *)arguments {
// 根据arguments数量选中SEL
SEL selector = [self selectorForArgumentCount:arguments.count];
// 根据SEL创建NSInvocation
NSInvocation *invocation = [NSInvocation invocationWithMethodSignature:[self methodSignatureForSelector:selector]];
invocation.selector = selector;
invocation.target = self;
for (NSUInteger i = 0; i < arguments.count; i++) {
id arg = arguments[i];
NSInteger argIndex = (NSInteger)(i + 2);
// 入参赋值
[invocation setArgument:&arg atIndex:argIndex];
}
// 方法调用
[invocation invoke];
__unsafe_unretained id returnVal;
// 获取调用返回值
[invocation getReturnValue:&returnVal];
return returnVal;
}
- (SEL)selectorForArgumentCount:(NSUInteger)count {
NSCParameterAssert(count > 0);
switch (count) {
case 0: return NULL;
case 1: return @selector(performWith:);
case 2: return @selector(performWith::);
case 3: return @selector(performWith:::);
case 4: return @selector(performWith::::);
case 5: return @selector(performWith:::::);
case 6: return @selector(performWith::::::);
case 7: return @selector(performWith:::::::);
case 8: return @selector(performWith::::::::);
case 9: return @selector(performWith:::::::::);
case 10: return @selector(performWith::::::::::);
case 11: return @selector(performWith:::::::::::);
case 12: return @selector(performWith::::::::::::);
case 13: return @selector(performWith:::::::::::::);
case 14: return @selector(performWith::::::::::::::);
case 15: return @selector(performWith:::::::::::::::);
}
NSCAssert(NO, @"The argument count is too damn high! Only blocks of up to 15 arguments are currently supported.");
return NULL;
}
// 具体实现
- (id)performWith:(id)obj1 {
id (^block)(id) = self.block;
return block(obj1);
}
- (id)performWith:(id)obj1 :(id)obj2 {
id (^block)(id, id) = self.block;
return block(obj1, obj2);
}
以此类推...
首先根据元组中元素的数量决定调用的SEL,然后动态创建NSInvocation,并调用他.
在给NSInvocation入参赋值是从i+2的位置开始给入参赋值,是因为前两个入参分别为id self和SEL _cmd.
在具体实现中是调用block,block的入参是元组的元素,返回值就是reduceBlock的返回值,由开发者返回.
startWith:
在第一个信号前插入一个信号.
实例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@1];
[subscriber sendNext:@2];
[subscriber sendNext:@3];
return [RACDisposable new];
}] startWith:@0];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"x: %@", [x stringValue]);
}];
输出:
2017-10-03 20:53:04.191798+0800 RAC[29045:1780844] x: 0
2017-10-03 20:53:04.191962+0800 RAC[29045:1780844] x: 1
2017-10-03 20:53:04.192030+0800 RAC[29045:1780844] x: 2
2017-10-03 20:53:04.192129+0800 RAC[29045:1780844] x: 3
源码:
- (__kindof RACStream *)startWith:(id)value {
return [[[self.class return:value]
concat:self]
setNameWithFormat:@"[%@] -startWith: %@", self.name, RACDescription(value)];
}
内部由concat方法实现,新建一个RACSignal直接返回value,然后concat原始的信号.前面说过concat的实现,所以信号stream的顺序是先发送value,然后发送原始信号.
skip:
跳过前n个信号.
示例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@1];
[subscriber sendNext:@2];
[subscriber sendNext:@3];
[subscriber sendNext:@4];
[subscriber sendNext:@5];
[subscriber sendNext:@6];
return [RACDisposable new];
}] skip:3];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"x: %@", [x stringValue]);
}];
输出:
2017-10-03 21:00:43.366280+0800 RAC[29217:1787626] x: 4
2017-10-03 21:00:43.366462+0800 RAC[29217:1787626] x: 5
2017-10-03 21:00:43.366585+0800 RAC[29217:1787626] x: 6
源码:
- (__kindof RACStream *)skip:(NSUInteger)skipCount {
Class class = self.class;
return [[self bind:^{
__block NSUInteger skipped = 0;
return ^(id value, BOOL *stop) {
if (skipped >= skipCount) return [class return:value];
skipped++;
return class.empty;
};
}] setNameWithFormat:@"[%@] -skip: %lu", self.name, (unsigned long)skipCount];
}
使用skipped记录跳过的数量,每忽略一次信号skipped+1,直到sikpped>=skipcount.
skipUntilBlock:
- (__kindof RACStream *)skipUntilBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
Class class = self.class;
return [[self bind:^{
__block BOOL skipping = YES;
return ^ id (id value, BOOL *stop) {
if (skipping) {
if (predicate(value)) {
skipping = NO;
} else {
return class.empty;
}
}
return [class return:value];
};
}] setNameWithFormat:@"[%@] -skipUntilBlock:", self.name];
}
skipUntilBlock:以predicate闭包作为筛选条件,当筛选条件为NO是跳过此信号,直到筛选条件为YES后面所有的信号都不跳过.
skipWhileBlock:
- (__kindof RACStream *)skipWhileBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
return [[self skipUntilBlock:^ BOOL (id x) {
return !predicate(x);
}] setNameWithFormat:@"[%@] -skipWhileBlock:", self.name];
}
skipWhileBlock:的信号集是skipUntilBlock:的信号集的补集。全集是原信号。skipWhileBlock:底层还是调用skipUntilBlock:，只不过判断条件的是不满足predicate( )闭包的集合。
take:
接受前n个信号.
示例代码:
RACSignal *signal = [[RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@1];
[subscriber sendNext:@2];
[subscriber sendNext:@3];
[subscriber sendNext:@4];
[subscriber sendNext:@5];
[subscriber sendNext:@6];
return [RACDisposable new];
}] take:3];
[signal subscribeNext:^(id _Nullable x) {
NSLog(@"x: %@", [x stringValue]);
}];
输出:
2017-10-03 21:12:43.774690+0800 RAC[29460:1801547] x: 1
2017-10-03 21:12:43.774919+0800 RAC[29460:1801547] x: 2
2017-10-03 21:12:43.775051+0800 RAC[29460:1801547] x: 3
源码:
- (__kindof RACStream *)take:(NSUInteger)count {
Class class = self.class;
if (count == 0) return class.empty;
return [[self bind:^{
__block NSUInteger taken = 0;
return ^ id (id value, BOOL *stop) {
if (taken < count) {
++taken;
if (taken == count) *stop = YES;
return [class return:value];
} else {
return nil;
}
};
}] setNameWithFormat:@"[%@] -take: %lu", self.name, (unsigned long)count];
}
takeUntilBlock:
- (__kindof RACStream *)takeUntilBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
Class class = self.class;
return [[self bind:^{
return ^ id (id value, BOOL *stop) {
if (predicate(value)) return nil;
return [class return:value];
};
}] setNameWithFormat:@"[%@] -takeUntilBlock:", self.name];
}
takeUntilBlock:是根据传入的predicate闭包作为筛选条件的。一旦predicate( )闭包满足条件，那么新信号停止发送新信号，因为它被置为nil了。和函数名的意思是一样的，take原信号的值，Until直到闭包满足条件。
takeWhileBlock:
- (__kindof RACStream *)takeWhileBlock:(BOOL (^)(id x))predicate {
NSCParameterAssert(predicate != nil);
return [[self takeUntilBlock:^ BOOL (id x) {
return !predicate(x);
}] setNameWithFormat:@"[%@] -takeWhileBlock:", self.name];
}
takeWhileBlock:的信号集是takeUntilBlock:的信号集的补集。全集是原信号。takeWhileBlock:底层还是调用takeUntilBlock:，只不过判断条件的是不满足predicate( )闭包的集合。
takeUntil:
- (RACSignal *)takeUntil:(RACSignal *)signalTrigger {
return [[RACSignal createSignal:^(id<RACSubscriber> subscriber) {
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
void (^triggerCompletion)(void) = ^{
[disposable dispose];
[subscriber sendCompleted];
};
RACDisposable *triggerDisposable = [signalTrigger subscribeNext:^(id _) {
triggerCompletion();
} completed:^{
triggerCompletion();
}];
[disposable addDisposable:triggerDisposable];
if (!disposable.disposed) {
RACDisposable *selfDisposable = [self subscribeNext:^(id x) {
[subscriber sendNext:x];
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{
[disposable dispose];
[subscriber sendCompleted];
}];
[disposable addDisposable:selfDisposable];
}
return disposable;
}] setNameWithFormat:@"[%@] -takeUntil: %@", self.name, signalTrigger];
}
当signalTrigger sendNext 或 sendCompleted时调用triggerCompletion闭包.阻断原信号.
takeUntilReplacement:
- (RACSignal *)takeUntilReplacement:(RACSignal *)replacement {
return [RACSignal createSignal:^(id<RACSubscriber> subscriber) {
RACSerialDisposable *selfDisposable = [[RACSerialDisposable alloc] init];
RACDisposable *replacementDisposable = [replacement subscribeNext:^(id x) {
[selfDisposable dispose];
[subscriber sendNext:x];
} error:^(NSError *error) {
[selfDisposable dispose];
[subscriber sendError:error];
} completed:^{
[selfDisposable dispose];
[subscriber sendCompleted];
}];
if (!selfDisposable.disposed) {
selfDisposable.disposable = [[self
concat:[RACSignal never]]
subscribe:subscriber];
}
return [RACDisposable disposableWithBlock:^{
[selfDisposable dispose];
[replacementDisposable dispose];
}];
}];
}
原始信号concat一个[RACSignal never]信号,这样可以保证原始信号完成不会调用新的信号的completed,可以一直等待
replacement信号.
当接收到replacement信号时,取消原来信号的订阅,由replacement信号代替原来的信号.
新的信号在没有接受到replacement信号时,信号由原始信号发送(不会发送sendCompleted信号),直到接收到replacement信号后,新的信号由replacement信号发送
+zip:
压缩多个信号,与-zip作业相同,不过-zip只能压缩俩个信号,+zip可以压缩多个信号.
示例代码:
RACSignal *signal1 = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@1];
[subscriber sendNext:@1];
return [[RACDisposable alloc] init];
}];
RACSignal *signal2 = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@2];
[subscriber sendNext:@1];
return [[RACDisposable alloc] init];
}];
RACSignal *signal3 = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@3];
[subscriber sendNext:@1];
return [[RACDisposable alloc] init];
}];
RACSignal *signal4 = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber> _Nonnull subscriber) {
[subscriber sendNext:@4];
[subscriber sendNext:@1];
return [[RACDisposable alloc] init];
}];
RACSignal *ziped = [RACSignal zip:RACTuplePack(signal1, signal2, signal3, signal4)];
[ziped subscribeNext:^(id _Nullable x) {
NSLog(@"x : %@", x);
}];
输出:
2017-10-04 11:24:58.416203+0800 RAC[2168:55396] x : <RACTuple: 0x60c0000197a0> (
1,
2,
3,
4
)
2017-10-04 11:24:58.416827+0800 RAC[2168:55396] x : <RACTuple: 0x604000019d60> (
1,
1,
1,
1
)
源码:
+ (__kindof RACStream *)zip:(id<NSFastEnumeration>)streams {
return [[self join:streams block:^(RACStream *left, RACStream *right) {
return [left zipWith:right];
}] setNameWithFormat:@"+zip: %@", streams];
}
+zip是对+join: block:方法的封装.
+ (__kindof RACStream *)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;
}
// 调用 外部block 关联两个block的逻辑关系
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];
}];
}
第一个信号的值用元组包裹,接着调用block依次zip后面的信号到current中.
此时信号中的每一个值由多层元组包裹,(((1), 2), 3) 像这样.
最后将多层元组重新整理,变成单层元组.
+zip:reduce
+zip:方法和reduceEach:方法的结合.
+ (__kindof RACStream *)zip:(id<NSFastEnumeration>)streams reduce:(id (^)())reduceBlock {
NSCParameterAssert(reduceBlock != nil);
RACStream *result = [self zip:streams];
// Although we assert this condition above, older versions of this method
// supported this argument being nil. Avoid crashing Release builds of
// apps that depended on that.
if (reduceBlock != nil) result = [result reduceEach:reduceBlock];
return [result setNameWithFormat:@"+zip: %@ reduce:", streams];
}
如果理解+zip和reduceEach的实现这个就很好理解了,这里就不在详细说明了.
scanWithStart: reduceWithIndex:
- (__kindof RACStream *)scanWithStart:(id)startingValue reduceWithIndex:(id (^)(id, id, NSUInteger))reduceBlock {
NSCParameterAssert(reduceBlock != nil);
Class class = self.class;
return [[self bind:^{
__block id running = startingValue;
__block NSUInteger index = 0;
return ^(id value, BOOL *stop) {
running = reduceBlock(running, value, index++);
return [class return:running];
};
}] setNameWithFormat:@"[%@] -scanWithStart: %@ reduceWithIndex:", self.name, RACDescription(startingValue)];
}
底层是由bind方法实现的,startingValue在block第一次调用的时候是running的值,running和next的逻辑关系由开发者自己实现,running总是指向block的返回值.
distinctUntilChanged
- (__kindof RACStream *)distinctUntilChanged {
Class class = self.class;
return [[self bind:^{
__block id lastValue = nil;
__block BOOL initial = YES;
return ^(id x, BOOL *stop) {
if (!initial && (lastValue == x || [x isEqual:lastValue])) return [class empty];
initial = NO;
lastValue = x;
return [class return:x];
};
}] setNameWithFormat:@"[%@] -distinctUntilChanged", self.name];
}
distinctUntilChanged的实现是用bind来完成的。每次变换中都记录一下原信号上一次发送过来的值，并与这一次进行比较，如果是相同的值，就“吞”掉，返回empty信号。只有和原信号上一次发送的值不同，变换后的新信号才把这个值发送出来。
