美文网首页
ReactiveCocoa-Signal

ReactiveCocoa-Signal

作者: 浅笑11 | 来源:发表于2018-01-30 23:46 被阅读0次

    ReactiveCocoa是一个将函数响应式编程(Functional Reactive Programming)带入到Objective-C中的开源库。

    RACSignal

    RAC中的核心是Signal,对应的类是RACSignal。它代表一个信号,Signal会给它的订阅者(Subscriber)发送一连串的数据,订阅者接到数据之后对这些数据进行处理。
    RACSignal被订阅的过程如下:

        RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
            
            [subscriber sendNext:@1];
            [subscriber sendNext:@2];
            [subscriber sendNext:@3];
            
            [subscriber sendCompleted];
            
            return [RACDisposable disposableWithBlock:^{
                NSLog(@"signal disposable");
            }];
        }];
        
        [signal subscribeNext:^(id  _Nullable x) {
            NSLog(@"value = %@",x);
        } error:^(NSError * _Nullable error) {
            NSLog(@"error = %@",error);
        } completed:^{
            NSLog(@"completed");
        }];
    
    • 调用RACSignal的createSignal:方法创建一个signal
    • 调用RACSignal的subscribeNext:方法,这个方法创建一个subscriber并调用subscribe:方法
    • subscribe:方法将subscriber传入didSubscribe的Block中
    • 在didSubscribe中通过[subscriber sendNext:]来执行next 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:"];
    }
    

    RACSignal的createSignal:方法通过调用子类RACDynamicSignald createSignal:方法返回一个Signal,并保存didSubscribe这个block。

    - (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
        NSCParameterAssert(nextBlock != NULL);
        NSCParameterAssert(errorBlock != NULL);
        NSCParameterAssert(completedBlock != NULL);
        
        RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
        return [self subscribe:o];
    }
    + (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;
    }
    

    subscribeNext:方法首先创建一个subscriber,并把next,error,completed三个block保存到subscriber中。接着调用RACDynamicSignal的subscribe:方法。

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

    subscribe:方法中执行保存在Signal中的didSubscribe,传入参数subscriber。

    - (void)sendNext:(id)value {
        @synchronized (self) {
            void (^nextBlock)(id) = [self.next copy];
            if (nextBlock == nil) return;
    
            nextBlock(value);
        }
    }
    

    didSubscribe中,传入的subscriber调用sendNext:方法执行保存在subscriber中的nextBlock

    通过上面的代码,可以发现ReactiveCocoa是通过block实现的订阅过程。signal中保存了didSubscribe,subscriber中保存了nextBlockerrorBlockcompletedBlockdidSubscribe的参数为一个subscriber,在block中,传入的subscriber调用sendNext:sendError:sendCompleted方法执行保存在subscriber中的block。didSubscribe的执行在signal的subscribe:方法中,而RAC将subscriber的初始化和subscribe:的调用放在subscribeNext:之中。

    按照上面的订阅过程,可以发现信号只有在被订阅的时候才会发送信号值,并且每订阅一次,didSubscribe就会被执行一次。这种信号称为冷信号。既然有冷信号,那么也有热信号,冷热信号的区别是:

    1. 热信号是主动的,尽管你并没有订阅事件,但是它会时刻推送,就像鼠标移动;而冷信号是被动的,只有当你订阅的时候,它才会发布消息。

    2. 热信号可以有多个订阅者,是一对多,集合可以与订阅者共享信息;而冷信号只能一对一,当有不同的订阅者,消息是重新完整发送。

    RACSubject

    在RAC中,冷信号基本都是RACSignal及其子类,而热信号则是RACSubject及其子类。

    @interface RACSubject<ValueType> : RACSignal<ValueType> <RACSubscriber>
    

    RACSubject继承RACSignal,并且遵守RACSubscriber协议,这说明既可以发送信号,也可以订阅信号。
    RACSubject的使用如下:

        RACSubject *subject = [RACSubject subject];
        
        [subject subscribeNext:^(id x) {
            NSLog(@"subscribe %@ ", x);
        }];
        [subject sendNext:@1];
        [subject sendNext:@2];
    

    接下来看一下RACSubject与RACSignal有什么不同

    - (instancetype)init {
        self = [super init];
        if (self == nil) return nil;
    
        _disposable = [RACCompoundDisposable compoundDisposable];
        _subscribers = [[NSMutableArray alloc] initWithCapacity:1];
        
        return self;
    }
    

    可以看到RACSubject有一个subscribers数组

    - (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
        NSCParameterAssert(subscriber != nil);
    
        RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
        subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
    
        NSMutableArray *subscribers = self.subscribers;
        @synchronized (subscribers) {
            [subscribers addObject:subscriber];
        }
        
        [disposable addDisposable:[RACDisposable disposableWithBlock:^{
            @synchronized (subscribers) {
                // Since newer subscribers are generally shorter-lived, search
                // starting from the end of the list.
                NSUInteger index = [subscribers indexOfObjectWithOptions:NSEnumerationReverse passingTest:^ BOOL (id<RACSubscriber> obj, NSUInteger index, BOOL *stop) {
                    return obj == subscriber;
                }];
    
                if (index != NSNotFound) [subscribers removeObjectAtIndex:index];
            }
        }]];
    
        return disposable;
    }
    

    通过subscribe:方法可以发现,subject的每次订阅都是将subscriber加到subscribers数组中。

    - (void)sendNext:(id)value {
        [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
            [subscriber sendNext:value];
        }];
    }
    

    从sendNext:的实现可以看出,每次RACSubject对象sendNext,都会对其中保留的subscribers进行sendNext,执行subscriber保存的nextBlock。

    RACSubject满足热信号的特点,它即使没有subscriber,因为自己继承了RACSubscriber协议,所以自己本身就可以发送信号。由于RACSubject将subscriber保存到数组中,所以可以给多个subscriber发送消息,并且subscriber只能接收到订阅之后发送的消息,订阅之前已经发出的消息则无法收到。

    RACSubject有一个子类RACReplaySubject,RACReplaySubject相对于RACSubject增加了一个数组用来保存发送过的数据,当有新的订阅时,会重发之前发送的数据。

    - (instancetype)initWithCapacity:(NSUInteger)capacity {
        self = [super init];
        
        _capacity = capacity;
        _valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
        
        return self;
    }
    - (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
        RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
    
        RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            @synchronized (self) {
                for (id value in self.valuesReceived) {
                    if (compoundDisposable.disposed) return;
    
                    [subscriber sendNext:(value == RACTupleNil.tupleNil ? nil : value)];
                }
    
                if (compoundDisposable.disposed) return;
    
                if (self.hasCompleted) {
                    [subscriber sendCompleted];
                } else if (self.hasError) {
                    [subscriber sendError:self.error];
                } else {
                    RACDisposable *subscriptionDisposable = [super subscribe:subscriber];
                    [compoundDisposable addDisposable:subscriptionDisposable];
                }
            }
        }];
    
        [compoundDisposable addDisposable:schedulingDisposable];
    
        return compoundDisposable;
    }
    

    冷信号转化为热信号

    由于RACSignal是冷信号,那么在用于网络请求或者是数据计算的时候,如果有多次订阅导致多次执行didSubscribe会发生重复请求以及数据错误。为了避免这种情况,可以使用RACMulticastConnection将冷信号转化为热信号。

    RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
            
            [subscriber sendNext:@1];
            [subscriber sendNext:@2];
            [subscriber sendNext:@3];
            
            [subscriber sendCompleted];
            
            return [RACDisposable disposableWithBlock:^{
                NSLog(@"signal disposable");
            }];
        }];
        
        RACMulticastConnection *connection = [signal multicast:[RACSubject subject]];
        
        [connection.signal subscribeNext:^(id  _Nullable x) {
            NSLog(@"value = %@",x);
        } error:^(NSError * _Nullable error) {
            NSLog(@"error = %@",error);
        } completed:^{
            NSLog(@"completed");
        }];
        
        [connection connect];
    

    对一个Signal进行multicast之后,我们是对connection.signal进行订阅而不是原来的signal。为什么connection.signal是热信号,我们看一下。

    - (RACMulticastConnection *)multicast:(RACSubject *)subject {
        [subject setNameWithFormat:@"[%@] -multicast: %@", self.name, subject.name];
        RACMulticastConnection *connection = [[RACMulticastConnection alloc] initWithSourceSignal:self subject:subject];
        return connection;
    }
    - (instancetype)initWithSourceSignal:(RACSignal *)source subject:(RACSubject *)subject {
        NSCParameterAssert(source != nil);
        NSCParameterAssert(subject != nil);
    
        self = [super init];
    
        _sourceSignal = source;
        _serialDisposable = [[RACSerialDisposable alloc] init];
        _signal = subject;
        
        return self;
    }
    

    通过RACMulticastConnection的初始化方法可以看出,RACMulticastConnection将原signal和传入的RACSubject保存,connection的signal实际是传入的RACSubject。

    - (RACDisposable *)connect {
        BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);
    
        if (shouldConnect) {
            self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
        }
    
        return self.serialDisposable;
    }
    

    [self.sourceSignal subscribe:_signal]会执行self.sourceSignal的didSubscribe这个block,并且将_signal当做订阅者。_signal在didSubscribe中会sendNext,这里的这个signal就是[RACSubject subject]。可以看出,一旦进入到这个didSubscribe中,后续的操作都是对这个[RACReplaySubject subject]进行的,与原来的sourceSignal彻底无关了。只有调用connect时,才执行sourceSignal的didSubscribe。这样我们不管订阅多少次connection.signal,只要connect只调用一次,那么didSubscribe也就只会执行一次。

    - (RACSignal *)autoconnect {
        __block volatile int32_t subscriberCount = 0;
    
        return [[RACSignal
            createSignal:^(id<RACSubscriber> subscriber) {
                OSAtomicIncrement32Barrier(&subscriberCount);
    
                RACDisposable *subscriptionDisposable = [self.signal subscribe:subscriber];
                RACDisposable *connectionDisposable = [self connect];
    
                return [RACDisposable disposableWithBlock:^{
                    [subscriptionDisposable dispose];
    
                    if (OSAtomicDecrement32Barrier(&subscriberCount) == 0) {
                        [connectionDisposable dispose];
                    }
                }];
            }]
            setNameWithFormat:@"[%@] -autoconnect", self.signal.name];
    }
    

    autoconnect通过订阅self.signal生成的信号,允许我们在第一次订阅connection.signal的时候自动调用connect

    除了multicast:方法,还有publishreplayreplayLastreplayLazily方法

    - (RACMulticastConnection *)publish {
        RACSubject *subject = [[RACSubject subject] setNameWithFormat:@"[%@] -publish", self.name];
        RACMulticastConnection *connection = [self multicast:subject];
        return connection;
    }
    

    publish方法只不过是去调用了multicast:方法,publish内部会新建好一个RACSubject,并把它当成入参传递给RACMulticastConnection。

    - (RACSignal *)replay {
        RACReplaySubject *subject = [[RACReplaySubject subject] setNameWithFormat:@"[%@] -replay", self.name];
    
        RACMulticastConnection *connection = [self multicast:subject];
        [connection connect];
    
        return connection.signal;
    }
    

    replay在创建RACMulticastConnection之后马上调用了connect方法,所以使用了RACReplaySubject,因为它会保存发送过的消息,当订阅之后依然会收到之前发送的消息,如果使用RACSubject会收不到消息。

    - (RACSignal *)replayLast {
        RACReplaySubject *subject = [[RACReplaySubject replaySubjectWithCapacity:1] setNameWithFormat:@"[%@] -replayLast", self.name];
    
        RACMulticastConnection *connection = [self multicast:subject];
        [connection connect];
    
        return connection.signal;
    }
    

    replayLastreplay的实现基本一样,不同的是replayLastCapacity值设为1,意味着只保留最新的值。

    - (RACSignal *)replayLazily {
        RACMulticastConnection *connection = [self multicast:[RACReplaySubject subject]];
        return [[RACSignal
            defer:^{
                [connection connect];
                return connection.signal;
            }]
            setNameWithFormat:@"[%@] -replayLazily", self.name];
    }
    

    replayLazilyconnect放在了defer中,当订阅信号的时候它才会被调用,如果含有和时间有关的操作,想要延迟执行,就可以用replayLazily

    相关文章

      网友评论

          本文标题:ReactiveCocoa-Signal

          本文链接:https://www.haomeiwen.com/subject/ssdvaxtx.html