改进RACCommand──一句话完成工作信号的创建、Next/Error事件的订阅，并保证互斥执行

iOS Blog

1.前言

本文将会介绍RACCommand进行异步操作（比如网络请求）的用法，分析其中存在的问题。然后介绍改进方案STButtonSignal的用法，给出STButtonSignal的原理以及具体实现。

同时推荐阅读我的另一篇文章——RAC扩展──异步filter、map

2.RACCommand的用法

假设我们有个下单按钮，点击之后提交新订单到服务器，一般我们会这么写。

self.submitBtn.rac_command =
[[RACCommand alloc] initWithEnabled:RACObserve(self.viewModel, agreeProtocol)
                        signalBlock:^RACSignal *(id input) {
                            return someWebRequestSignal;
}];
[[self.submitBtn.rac_command.executionSignals flatten] subscribeNext:^(id value) {
    //Handle data from server
}];
[self.submitBtn.rac_command.errors subscribeNext:^(NSError *error) {
    //Handle error
}];

这样使用存在如下问题：
1. 网络请求信号的创建和订阅过程是分离的，代码逻辑不够集中。
2. 网络请求的Next事件、Error事件是分离的，为了处理它们，我需要订阅两个信号。
3. Next事件信号executionSignals是signal of signals，想要订阅网络请求返回数据要先进行flatten操作，这个操作是重复的，产生了门板代码。

3. RACCommand为什么需要这么用

1.为什么工作信号的创建和订阅要分离？
因为，在RACCommand实现中subscriber并不直接订阅工作信号。按钮是可以重复点击的，每次点击可以根据输入（input）创建不同的工作信号，RACCommand会在“创建”与“订阅”中间插入一些对工作信号的变换处理操作，实现控制并发、持久订阅等功能，详细可以参考源码。

2.为什么工作信号的Next事件和Error事件要分开从executionSignals、errors两个信号发出，我直接订阅一个Signal不是更简单吗？
因为订阅关系会在Erro发出的时候自动解除。关键源码如下（RACSubscriber.m）：

- (void)sendError:(NSError *)e {
    @synchronized (self) {
        void (^errorBlock)(NSError *) = [self.error copy];
        [self.disposable dispose];
        if (errorBlock == nil) return;
        errorBlock(e);
    }
}

假如让Next事件和Error事件从同一个信号发出，如果我点击按钮，进行网络请求，出错了，订阅关系自动解除。那么我再次点击按钮，响应处理代码就不会被执行。另外我们注意到errors信号内发送的也是封装过的error事件，所以可以持续接收Error事件。

4.RACommand的性能问题

首先我们知道，我们拿到的executionSignals是Signal of signals，读源码我们可以发现这个executionSignals是一个signal通过concat得到的。关键源码（RACCommand.m）：

RACSignal *newActiveExecutionSignals = [[[[[self
    rac_valuesAndChangesForKeyPath:@keypath(self.activeExecutionSignals) options:NSKeyValueObservingOptionNew observer:nil]
    reduceEach:^(id _, NSDictionary *change) {
        NSArray *signals = change[NSKeyValueChangeNewKey];
        if (signals == nil) return [RACSignal empty];
        return [signals.rac_sequence signalWithScheduler:RACScheduler.immediateScheduler];
    }]
    concat]
    publish]
    autoconnect];
_executionSignals = [[[newActiveExecutionSignals
    map:^(RACSignal *signal) {
        return [signal catchTo:[RACSignal empty]];
    }]
    deliverOn:RACScheduler.mainThreadScheduler]
    setNameWithFormat:@"%@ -executionSignals", self];

那这个signal就是signal of (signal of signal)s，一个三维的signal，好复杂。分析一下，这个signal是由self.activeExecutionSignals产生的，而self.activeExecutionSignals是一个signal数组，保存的是正在执行的工作信号。self.activeExecutionSignals之所以是一个数组，而不是单个signal，是为了支持并发执行工作信号。
也就是说，RACCommand为了支持并发，内部采用了一种比较复杂的实现。而我们绝大多数的应用场景中，按钮的处理逻辑是互斥的，我们完全可以采用另外一种比较简单的实现，不去支持并发，获取更高的性能。
针对以上所述的RACCommand的用法复杂和性能的问题，我进行改进，下面介绍一下我改进的STButtonSignal的用法和实现原理。

5. STButtonSignal的用法

用法示例：

    [[STButtonSignal associateButton:self.submitButton
                                withSignalBlock:^RACSignal *(id input) {
                                    return  someWebRequestSignal;
                                }]
                                subscribeNext:^(id x) {
                                    //Handle data from server
                                } error:^(NSError *error) {
                                   //Handle error
                               }];

注意：someWebRequestSignal网络请求返回之后一定要发出Complete事件，如果你对someWebRequestSignal进行了flattenMap
，flattenMap出来的Signal也一定要发出Complete事件，我依赖complete事件将Button恢复为enable状态，如果不发送compete事件，Button将一直处于disable状态。

这段代码看着问题很多。比如：
1. 你只订阅了一个Signal，可是我的按钮是可以重复点击的，新创建workSignal没有被订阅，这怎么行？
2. 我一个workSignal sendComplete或者sendError了，你的订阅关系不就结束了，这个按钮不就没法点击了吗？
别着急，我内部做了处理，这些问题都没问题。

6. STButtonSignal主要解决的问题以及解决办法

1.并发问题（比如，快速点击按钮导致的重复提交订单问题）
按照如下流程控制按钮状态，按钮被点击->disable按钮->创建workSignal->workSignal complete->enable按钮。

2.我们知道Button可重复点击（在上一次响应处理结束后），每次点击都将创建一个Signal，我们要让一个subscriber订阅这些Signal。
借鉴Multicast的实现原理，借用一个RACSubject中中间做消息转发，调用方的subscriber实际将会订阅这个RACSubject，这个RACSubject将会订阅workSignal。如下图所示：

3.subscriber可以持续接收error、complete事件，并不会因为接收到error、complete事件而终止订阅。
自定义一个STManualDisposeSubscriber，它和RACSubscriber唯一的不同是不会在接收到error、complete事件时自动dispose。然后，STButtonSignal复写了RACSignal中订阅方法，获取调用方的subscriber，在内部转化成为一个我自定义的STManualDisposeSubscriber。自定义subscriber不会因为接收到error、complete事件而自动dispose掉自己，所以一次订阅每次按钮点击处理的结果（无论成功还是失败）都能接收到。

7. 完整源码

#import "STButtonSignal.h"
#import "STManualDisposeSubscriber.h"
static void *UIButtonSignalKey = &UIButtonSignalKey;
@interface STButtonSignal ()
@property (nonatomic, strong, readonly) RACSignal * (^signalBlock)(id input);
@property (nonatomic, strong, readonly) UIButton *button;
@property (nonatomic, strong) RACDisposable *activeSignalDisposable;
@property (nonatomic, strong) RACSubject *transitSubject;
@end
@implementation STButtonSignal
+ (STButtonSignal *)associateButton:(UIButton *)button withSignalBlock:(RACSignal * (^)(id input))signalBlock{
    STButtonSignal *signal = [[STButtonSignal alloc] initWithSignalBlock:(RACSignal * (^)(id input))signalBlock button:(UIButton *)button];
    return signal;
}
+ (STButtonSignal *)createSignalWithSignalBlock:(RACSignal * (^)(id input))signalBlock button:(UIButton *)button {
    STButtonSignal *signal = [[STButtonSignal alloc] initWithSignalBlock:(RACSignal * (^)(id input))signalBlock button:(UIButton *)button];
    return signal;
}
- (instancetype)initWithSignalBlock:(RACSignal * (^)(id input))signalBlock button:(UIButton *)button{
    self = [super init];
    if (self) {
        _signalBlock = signalBlock;
        _button = button;
        _transitSubject = [[RACSubject alloc] init];
        [self rac_hijackActionAndTargetIfNeeded];
        objc_setAssociatedObject(button, UIButtonSignalKey, self, OBJC_ASSOCIATION_RETAIN_NONATOMIC);
    }
    return self;
}
- (void)rac_hijackActionAndTargetIfNeeded {
    SEL hijackSelector = @selector(rac_commandPerformAction:);
    for (NSString *selector in [self.button actionsForTarget:self forControlEvent:UIControlEventTouchUpInside]) {
        if (hijackSelector == NSSelectorFromString(selector)) {
            return;
        }
    }
    [self.button addTarget:self action:hijackSelector forControlEvents:UIControlEventTouchUpInside];
}
- (void)rac_commandPerformAction:(id)sender {
//    [self.rac_command execute:sender];
    NSAssert(self.activeSignalDisposable == nil || [self.activeSignalDisposable isDisposed], @"Don't allow concurrent execution, activeSignal should complete and be set nil when you can click the button again");
    RACSignal *newWorkSignal = self.signalBlock(sender);
    self.activeSignalDisposable = [[[newWorkSignal initially:^{
                                                    [self.button setEnabled:NO];
                                                }] finally:^{
                                                    [self.button setEnabled:YES];
                                                }] subscribeNext:^(id x) {
                                                    [self.transitSubject sendNext:x];
                                                } error:^(NSError *error) {
                                                    [self.transitSubject sendError:error];
                                                } completed:^{
                                                    [self.transitSubject sendCompleted];
                                                }];
}
#pragma mark RACSubscriber
//Override subscribe method, use STManualDisposeSubscriber to replace RACSubscriber
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCAssert(NO, @"This method is not implemented yet");
    return nil;
}
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock completed:(void (^)(void))completedBlock {
    NSCParameterAssert(nextBlock != NULL);
    NSCParameterAssert(completedBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:nextBlock error:NULL completed:completedBlock];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
    NSCParameterAssert(nextBlock != NULL);
    NSCParameterAssert(errorBlock != NULL);
    NSCParameterAssert(completedBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeError:(void (^)(NSError *error))errorBlock {
    NSCParameterAssert(errorBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:NULL error:errorBlock completed:NULL];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeCompleted:(void (^)(void))completedBlock {
    NSCParameterAssert(completedBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:NULL error:NULL completed:completedBlock];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock {
    NSCParameterAssert(nextBlock != NULL);
    NSCParameterAssert(errorBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:nextBlock error:errorBlock completed:NULL];
    return [self.transitSubject subscribe:o];
}
- (RACDisposable *)subscribeError:(void (^)(NSError *))errorBlock completed:(void (^)(void))completedBlock {
    NSCParameterAssert(completedBlock != NULL);
    NSCParameterAssert(errorBlock != NULL);
    STManualDisposeSubscriber *o = [STManualDisposeSubscriber subscriberWithNext:NULL error:errorBlock completed:completedBlock];
    return [self.transitSubject subscribe:o];
}
@end

#import "STManualDisposeSubscriber.h"
#import "RACSubscriber.h"
#import "RACEXTScope.h"
#import "RACCompoundDisposable.h"
@interface STManualDisposeSubscriber ()
// These callbacks should only be accessed while synchronized on self.
@property (nonatomic, copy) void (^next)(id value);
@property (nonatomic, copy) void (^error)(NSError *error);
@property (nonatomic, copy) void (^completed)(void);
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;
@end
@implementation STManualDisposeSubscriber
#pragma mark Lifecycle
+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
    STManualDisposeSubscriber *subscriber = [[STManualDisposeSubscriber alloc] init];
    subscriber->_next = [next copy];
    subscriber->_error = [error copy];
    subscriber->_completed = [completed copy];
    return subscriber;
}
- (id)init {
    self = [super init];
    if (self == nil) return nil;
    @unsafeify(self);
    RACDisposable *selfDisposable = [RACDisposable disposableWithBlock:^{
        @strongify(self);
        @synchronized (self) {
            self.next = nil;
            self.error = nil;
            self.completed = nil;
        }
    }];
    _disposable = [RACCompoundDisposable compoundDisposable];
    [_disposable addDisposable:selfDisposable];
    return self;
}
- (void)dealloc {
    [self.disposable dispose];
}
#pragma mark RACSubscriber
- (void)sendNext:(id)value {
    @synchronized (self) {
        void (^nextBlock)(id) = [self.next copy];
        if (nextBlock == nil) return;
        nextBlock(value);
    }
}
- (void)sendError:(NSError *)e {
    @synchronized (self) {
        void (^errorBlock)(NSError *) = [self.error copy];
//        [self.disposable dispose];
        if (errorBlock == nil) return;
        errorBlock(e);
    }
}
- (void)sendCompleted {
    @synchronized (self) {
        void (^completedBlock)(void) = [self.completed copy];
//        [self.disposable dispose];
        if (completedBlock == nil) return;
        completedBlock();
    }
}
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)otherDisposable {
    if (otherDisposable.disposed) return;
    RACCompoundDisposable *selfDisposable = self.disposable;
    [selfDisposable addDisposable:otherDisposable];
    @unsafeify(otherDisposable);
    // If this subscription terminates, purge its disposable to avoid unbounded
    // memory growth.
    [otherDisposable addDisposable:[RACDisposable disposableWithBlock:^{
        @strongify(otherDisposable);
        [selfDisposable removeDisposable:otherDisposable];
    }]];
}
@end