ReactiveCocoa信号组合操作merge和zip

1. merge

merge的源代码进行转换后如下所示：

+ (RACSignal *)merge:(id<NSFastEnumeration>)signals {
    NSMutableArray *copiedSignals = [[NSMutableArray alloc] init];
    for (RACSignal *signal in signals) {
        [copiedSignals addObject:signal];
    }
    
    RACSignal *tempSignal = [RACSignal createSignal:^ RACDisposable * (id<RACSubscriber> subscriber) {
        for (RACSignal *signal in copiedSignals) {
            [subscriber sendNext:signal];
        }
        
        [subscriber sendCompleted];
        return nil;
    }];
    
    RACSignal *rtSignal = [tempSignal flattenMap:^(id value) {
        NSCAssert([value isKindOfClass:RACStream.class], @"Stream %@ being flattened contains an object that is not a stream: %@", stream, value);
        return value;
    }];
    
    return rtSignal;
}

1. tempSignal这个信号的didSubscribe仅仅是将信号数组中的信号signal_i一个个传递出去；
2. 对tempSignal进行flattenMap转换仅仅是将tempSignal传递的值signal_i直接传递到下一步，没有额外操作。
3. 在订阅rtSignal时，rtSignal的didSubscribe会将flattenMap传递出来的每一个signal_i进行订阅，将结果传递出去。

其实merge的作用就如同它的名字一样，是将多个信号合并成一个信号；对合并后的信号进行订阅时，原来的信号都会被订阅，而这些信号在sendNext时也都会执行同一个next。

2. zip

/// Zips the values in the given streams to create RACTuples.
///
/// The first value of each stream will be combined, then the second value, and
/// so forth, until at least one of the streams is exhausted.
///
/// streams - The streams to combine. These must all be instances of the same
///           concrete class implementing the protocol. If this collection is
///           empty, the returned stream will be empty.
///
/// Returns a new stream containing RACTuples of the zipped values from the
/// streams.
+ (instancetype)zip:(id<NSFastEnumeration>)streams;

先翻译一下zip的作用，给定一个信号数组signal_array[N]，创建一个信号zip_return，当订阅zip_return时，会等待signal_array中每一个信号都sendNext:value_i后，zip_return才会sendNext，zip_return传出的值是[value₁,...,value_N]。

看一下zip的源码：

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

上面代码涉及到join和zipWith两个方法，先看zipWith方法。

2.1 zipWith

RACSignal.m

- (instancetype)zipWith:(RACStream *)stream

1. zipWith返回一个信号zipWith_return_signal；
2. 在订阅zipWith_return_signal时，当且仅当当前信号self和传入的参数信号signal都sendNext值时，才会将获取到的值(以一个tuple形式，[value_self，value_signal]，self信号传出的值在前，signal信号传出的值在后)吐出去；
3. 当前信号self或入参信号signal发出complete时，zipWith_return_signal"都会"sendCompleted.

zipWith有个坑，就是zipWith的两个信号sendNext的数目以sendNext数目最少的信号为准，什么意思呢，看个具体例子：

- (void)viewDidLoad {
    [super viewDidLoad];
    // Do any additional setup after loading the view, typically from a nib.
    
    
    RACSignal *zippedSignal = [RACSignal zip:@[[self fetchData1],
                                               [self fetchData3]]];
    
     [zippedSignal subscribeNext:^(RACTuple *tuple) {
         NSLog(@"%@", tuple);
     }];
    
}

- (RACSignal *)fetchData1 {
    return [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        [self httpRequest:@"Post" param:@{@"commandKey":@"request1"} completion:^(id response) {
            [subscriber sendNext:@11];
            [subscriber sendNext:@22];
            [subscriber sendCompleted];
        }];
        return nil;
    }];
}

- (RACSignal *)fetchData3 {
    return [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        [self httpRequest:@"Post" param:@{@"commandKey":@"request3"} completion:^(id response) {
            [subscriber sendNext:@33];
            [subscriber sendCompleted];
        }];
        return nil;
    }];
}

上面fetchData1中sendNext了2次，fetchData3sendNext了1次，最终执行结果是(11,33)，也就是fetchData3在sendNext:@33后，然后再sendCompleted后zipWith就完成了，不会去理会fetchData1还有sendNext:@22。因此，zip或者zipWith涉及到的信号劲量保持sendNext数目一致。

所以zipWith的作用就是控制2个信号一起返回，那么如果想要控制多个信号一起返回该怎么做呢，下面看一下join方法。

2.2 join

RACStream.m

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

1. for循环部分，如果信号数组里只有1个信号，那么通过map，最终获取的结果被放进一个tuple里，比如[value1]；如果数组里信号多余1个，那么第一个信号和第二信号就要做zipWith(此时block就是zipWith)操作，参考上面zipWith，得到的返回结果是[[value1]，value2]；如果还有第3个信号，那么将信号1和信号2 zipWith的结果与信号3继续zipWith，得到的结果就是[[[value1]，value2], value3];
2. return部分，for循环得到的信号最后sendNext的值是一个[[[value1]，value2], value3]之类的tuple，如同注释里的"(((1), 2), 3)"一样，需要转换成[value1, value2, value3]。