ReactiveCocoa学习笔记<三> RACSignal基本

作者: 樂亦leeyii | 来源:发表于2017-10-04 16:28 被阅读0次

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信号。只有和原信号上一次发送的值不同，变换后的新信号才把这个值发送出来。

ReactiveCocoa学习笔记<三> RACSignal基本

filter: 过滤

RACReturnSignal

RACEmptySignal

ignore: 忽略

reduceEach:

startWith:

skip:

skipUntilBlock:

skipWhileBlock:

take:

takeUntilBlock:

takeWhileBlock:

takeUntil:

takeUntilReplacement:

+zip:

+zip:reduce

scanWithStart: reduceWithIndex:

distinctUntilChanged

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读