ReactiveObjC - RACStream

抽象方法

RACStream中的许多定义都是抽象的，没有具体实现，需要由其子类进行实现。

empty

+ (__kindof RACStream *)empty;

因为RAC中nil会导致crash，所以很多时候需要定义一个空对象来替代nil，一般empty都被创建为一个单例来使用。

bind

- (__kindof RACStream *)bind:(RACStreamBindBlock (^)(void))block;

懒绑定，将block返回的RACStream绑定到自身，在调用时才执行内部操作。很多操作都是基于bind的，比如flattenMap，大多时候我们都调用bind的上层方法。

return

+ (__kindof RACStream *)return:(id)value;

把一个值包装成对应的RACStream的子类型。

concat

- (__kindof RACStream *)concat:(RACStream *)stream;

连接两个信号，子类实现具体如何连接。

zipWith

- (__kindof RACStream *)zipWith:(RACStream *)stream;

压缩两个信号，子类实现具体如何压缩。

---

RACStream的一些操作

下面是一些RACStream中常用的有具体实现的操作方法。

flattenMap、map、flatten、mapReplace

- (__kindof RACStream *)flattenMap:(__kindof RACStream * (^)(id value))block {
    Class class = self.class;

    return [[self bind:^{
        return ^(id value, BOOL *stop) {
            id stream = block(value) ?: [class empty];
            NSCAssert([stream isKindOfClass:RACStream.class], @"Value returned from -flattenMap: is not a stream: %@", stream);

            return stream;
        };
    }] setNameWithFormat:@"[%@] -flattenMap:", self.name];
}

以下三者都是基于flattenMap，对RACStream中的值进行映射
- (__kindof RACStream *)flatten;
- (__kindof RACStream *)map:(id (^)(id value))block;
- (__kindof RACStream *)mapReplace:(id)object;

combinePreviousWithStart：reduce：

- (__kindof RACStream *)combinePreviousWithStart:(id)start reduce:(id (^)(id previous, id next))reduceBlock {
    NSCParameterAssert(reduceBlock != NULL);
    return [[[self
        scanWithStart:RACTuplePack(start)
        reduce:^(RACTuple *previousTuple, id next) {
            id value = reduceBlock(previousTuple[0], next);
            return RACTuplePack(next, value);
        }]
        map:^(RACTuple *tuple) {
            return tuple[1];
        }]
        setNameWithFormat:@"[%@] -combinePreviousWithStart: %@ reduce:", self.name, RACDescription(start)];
}

将传入的start作为初值，按下标两两计算返回当前下标对应的新值，返回一个新的信号。

filter

- (__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];
}

按照block的返回布尔值来过滤掉信号中的值。

ignore

- (__kindof RACStream *)ignore:(id)value {
    return [[self filter:^ BOOL (id innerValue) {
        return innerValue != value && ![innerValue isEqual:value];
    }] setNameWithFormat:@"[%@] -ignore: %@", self.name, RACDescription(value)];
}

相当于按照ignore的参数为过滤条件来过滤。

reduceEach

- (__kindof RACStream *)reduceEach:(RACReduceBlock)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];
}

当信号值为RACTuple时，通过block中的逻辑将RACTuple转换一个新的值返回。

startWith

- (__kindof RACStream *)startWith:(id)value {
    return [[[self.class return:value]
        concat:self]
        setNameWithFormat:@"[%@] -startWith: %@", self.name, RACDescription(value)];
}

把value按照当前类型封装，组装到信号的开头，成为一个新信号返回。

skip

- (__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];
}

跳过信号中的前skipCount个值。

take

- (__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];
}

返回信号中前count个值。

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

        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];
    }];
}

把参数streams这一组信号，打包成一个信号返回，打包方式是在block中两两逐个打包。

zip：

+ (__kindof RACStream *)zip:(id<NSFastEnumeration>)streams {
    return [[self join:streams block:^(RACStream *left, RACStream *right) {
        return [left zipWith:right];
    }] setNameWithFormat:@"+zip: %@", streams];
}

底层是join：block：，打包逻辑是信号两两进行zipWith操作。

zip：reduce：

+ (__kindof RACStream *)zip:(id<NSFastEnumeration>)streams reduce:(RACGenericReduceBlock)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操作得到的信号进行reduce操作。

cancat

+ (__kindof RACStream *)concat:(id<NSFastEnumeration>)streams {
    RACStream *result = self.empty;
    for (RACStream *stream in streams) {
        result = [result concat:stream];
    }

    return [result setNameWithFormat:@"+concat: %@", streams];
}

将传入的一组信号进行连接。

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)];
}

扫描信号，startingValue对应block中一个参数，block返回值对应下一值扫描的startingValue。

scanWithStart：reduce：

- (__kindof RACStream *)scanWithStart:(id)startingValue reduce:(id (^)(id running, id next))reduceBlock {
    NSCParameterAssert(reduceBlock != nil);

    return [[self
        scanWithStart:startingValue
        reduceWithIndex:^(id running, id next, NSUInteger index) {
            return reduceBlock(running, next);
        }]
        setNameWithFormat:@"[%@] -scanWithStart: %@ reduce:", self.name, RACDescription(startingValue)];
}

同scanWithStart：reduceWithIndex：，只是忽略了当前扫描值的下标。

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];
}

返回所有值，除非predicate返回值为YES；

takeWhileBlock

- (__kindof RACStream *)takeWhileBlock:(BOOL (^)(id x))predicate {
    NSCParameterAssert(predicate != nil);

    return [[self takeUntilBlock:^ BOOL (id x) {
        return !predicate(x);
    }] setNameWithFormat:@"[%@] -takeWhileBlock:", self.name];
}

相反的，返回所有值，除非predicate返回值为NO；

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];
}

跳过所有值，除非predicate返回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];
}

相反的，跳过所有值，除非predicate返回值为NO；

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];
}

将连续的相同的值合并为一个值。