本文是投稿文章，作者：聽榆大叔（博客） 

---

開車不需要知道離合器是怎麼工作的，但如果知道離合器原理，那麼車子可以開得更平穩。

ReactiveCocoa 是一個重型的 FRP 框架，內容十分豐富，它使用了大量內建的 block，這使得其有強大的功能的同時，內部源碼也比較復雜。本文研究的版本是2.4.4，小版本間的差別不是太大，無需擔心此問題。 這裡只探究其核心 RACSignal 源碼及其相關部分。本文不會詳細解釋裡面的代碼，重點在於討論那些核心代碼是怎麼來的。文本難免有不正確的地方，請不吝指教，非常感謝。

@protocol RACSubscriber

信號是一個異步數據流，即一個將要發生的以時間為序的事件序列，它能發射出三種不同的東西：value、error、completed。咱們能異步地捕獲這些事件：監聽信號，針對其發出的三種東西進行操作。“監聽”信息的行為叫做 訂閱(subscriber)。我們定義的操作就是觀察者，這個被“監聽”的信號就是被觀察的主體(subject) 。其實，這正是“觀察者”設計模式！

RAC 針對這個訂閱行為定義了一個協議：RACSubscriber。RACSubscriber 協議是與 RACSignal 打交道的唯一方式。咱們先不探究 RACSignal 的內容，而是先研究下 RACSubscriber 是怎麼回事。

先來看下 RACSubscriber 的定義：

// 用於從 RACSignal 中直接接收 values 的對象
@protocol RACSubscriber
@required

/// 發送下一個 value 給 subscribers。value 可以為 nil。
- (void)sendNext:(id)value;

/// 發送 error 給 subscribers。 error 可以為 nil。
///
/// 這會終結整個訂閱行為，而且接下來也無法再訂閱任何信號了。
- (void)sendError:(NSError *)error;

/// 發送 completed 給 subscribers。
///
/// 這會終結整個訂閱行為，而且接下來也無法再訂閱任何信號了。
- (void)sendCompleted;

/// 現在重要的是上面三個，先別管這個，忽略掉。
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable;

@end

1、NLSubscriber

咱們自己來實現這個協議看看（本文自定義的類都以 “NL” 開頭，以視區別）：

// NLSubscriber.h
@interface NLSubscriber : NSObject @end
// NLSubscriber.m
@implementation NLSubscriber
- (void)sendNext:(id)value {
  NSLog(@"%s value:%@", sel_getName(_cmd), value);
}
- (void)sendCompleted {
  NSLog(@"%s", sel_getName(_cmd));
}
- (void)sendError:(NSError *)error {
  NSLog(@"%s error:%@", sel_getName(_cmd), error);
}
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable {
  // to nothing
}
@end

現在咱們這個類只關心 sendNext: 、 sendError: 和 sendCompleted。本類的實現只是簡單的打印一些數據。那怎麼來使用這個訂閱者呢？RACSignal 類提供了接口來讓實現了 RACSubscriber 協議的訂閱者訂閱信號：

@interface RACSignal (Subscription)
/*
*  `subscriber` 訂閱 receiver 的變化。由 receiver 決定怎麼給 subscriber 發送事件。
*簡單來說，就是由這個被訂閱的信號來給訂閱者 subscriber 發送 `sendNext:` 等消息。
*/
- (RACDisposable *)subscribe:(id)subscriber;
@end

用定時器信號來試試看：

/**
*  @brief  創建一個定時器信號，每三秒發出一個當時日期值。一共發5次。
*/
RACSignal *signalInterval = [RACSignal interval:3.0 onScheduler:[RACScheduler mainThreadScheduler]];
signalInterval = [signalInterval take:5];
NLSubscriber *subscriber = [[NLSubscriber alloc] init];

/**
*  @brief  用訂閱者 subscriber 訂閱定時器信號
*/
[signalInterval subscribe:subscriber];

下面是輸出結果：

2015-08-15 17:45:02.612 RACPraiseDemo[738:59818] sendNext: value:2015-08-15 09:45:02 +0000
2015-08-15 17:45:05.612 RACPraiseDemo[738:59818] sendNext: value:2015-08-15 09:45:05 +0000
2015-08-15 17:45:08.615 RACPraiseDemo[738:59818] sendNext: value:2015-08-15 09:45:08 +0000
2015-08-15 17:45:11.613 RACPraiseDemo[738:59818] sendNext: value:2015-08-15 09:45:11 +0000
2015-08-15 17:45:14.615 RACPraiseDemo[738:59818] sendNext: value:2015-08-15 09:45:14 +0000
2015-08-15 17:45:14.615 RACPraiseDemo[738:59818] sendCompleted

2、改進NLSubscriber

現在的這個訂閱者類 NLSubscriber 除了打印打東西外，啥也干不了，更別說復用了，如果針對所有的信號都寫一個訂閱者那也太痛苦了，甚至是不太可能的事。

咱們來改進一下，做到如下幾點：

a.實現 RACSubscriber 協議

b.提供與 RACSubscriber 相 對應的 、可選的 、可配的接口。

沒錯，這正是一個適配器！

第2點的要求可不少，那怎麼才能做到這一點呢？還好，OC 中有 block ！咱們可以將 RACSubscriber 協議中的三個方法轉為三個 block：

- (void)sendNext:(id)value;           ---->    void (^next)(id value);
- (void)sendError:(NSError *)error;   ---->    void (^error)(NSError *error);
- (void)sendCompleted;                ---->    void (^completed)(void);

改進目標和改進方向都有了，那咱們來看看改進後的的樣子：

// 頭文件
/**
 *  @brief  基於 block 的訂閱者
 */
@interface NLSubscriber : NSObject /**
 *  @brief  創建實例
 */
+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed;
@end
// 實現文件
@interface NLSubscriber ()
@property (nonatomic, copy) void (^next)(id value);
@property (nonatomic, copy) void (^error)(NSError *error);
@property (nonatomic, copy) void (^completed)(void);
@end
@implementation NLSubscriber
#pragma mark Lifecycle
+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
  NLSubscriber *subscriber = [[self alloc] init];
  subscriber->_next = [next copy];
  subscriber->_error = [error copy];
  subscriber->_completed = [completed copy];
  return subscriber;
}
#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];
    if (errorBlock == nil) return;
    errorBlock(e);
  }
}
- (void)sendCompleted {
  @synchronized (self) {
    void (^completedBlock)(void) = [self.completed copy];
    if (completedBlock == nil) return;
    completedBlock();
  }
}
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable {
  // to nothing
}
@end

現在來試試看這個改進版，還是上面那個定時器的例子：

/**
   *  @brief  創建一個定時器信號，每三秒發出一個當時日期值。一共發5次。
   */
  RACSignal *signalInterval = [RACSignal interval:3.0 onScheduler:[RACScheduler mainThreadScheduler]];
  signalInterval = [signalInterval take:5];
  NLSubscriber *subscriber = [NLSubscriber subscriberWithNext:^(id x) {
    NSLog(@"next:%@", x);
  } error:nil completed:^{
    NSLog(@"completed");
  }];
  /**
   *  @brief  用訂閱者 subscriber 訂閱定時器信號
   */
  [signalInterval subscribe:subscriber];

輸出結果如下：

2015-08-15 19:50:43.355 RACPraiseDemo[870:116551] next:2015-08-15 11:50:43 +0000
2015-08-15 19:50:46.358 RACPraiseDemo[870:116551] next:2015-08-15 11:50:46 +0000
2015-08-15 19:50:49.355 RACPraiseDemo[870:116551] next:2015-08-15 11:50:49 +0000
2015-08-15 19:50:52.356 RACPraiseDemo[870:116551] next:2015-08-15 11:50:52 +0000
2015-08-15 19:50:55.356 RACPraiseDemo[870:116551] next:2015-08-15 11:50:55 +0000
2015-08-15 19:50:55.356 RACPraiseDemo[870:116551] completed

輸出結果沒什麼變化，但是訂閱者的行為終於受到咱們的撐控了。再也不用為了一個信號而去實現 RACSubscriber 協議了，只需要拿出 NLSubscriber 這個適配器，再加上咱們想要的自定義的行為即可。如果對信號發出的某個事件不感興趣，直接傳個 nil 可以了，例如上面例子的 error: ，要知道， RACSubscriber 協議中的所有方法都是 @required 的。NLSubscriber 大大方便了我們的工作。

那還以再改進嗎？

3、RACSignal 類別之 Subscription

有沒有可能把 NLSubscriber 隱藏起來呢？畢竟作為一個信號的消費者，需要了解的越少就越簡單，用起來也就越方便。咱們可以通過 OC 中的類別方式，給 RACSignal 加個類別（nl_Subscription），將訂閱操作封裝到這個信號類中。這樣，對於使用這個類的客戶而言，甚至不知道訂閱者的存在。

nl_Subscription 類別代碼如下：

// .h 
#import "RACSignal.h"
@interface RACSignal (nl_Subscription)
- (void)nl_subscribeNext:(void (^)(id x))nextBlock;
- (void)nl_subscribeNext:(void (^)(id x))nextBlock completed:(void (^)(void))completedBlock;
- (void)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock;
- (void)nl_subscribeError:(void (^)(NSError *error))errorBlock;
- (void)nl_subscribeCompleted:(void (^)(void))completedBlock;
- (void)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock;
- (void)nl_subscribeError:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock;
@end
// .m
#import "RACSignal+nl_Subscription.h"
#import "NLSubscriber.h"
@implementation RACSignal (nl_Subscription)
- (void)nl_subscribeNext:(void (^)(id x))nextBlock {
  [self nl_subscribeNext:nextBlock error:nil completed:nil];
}
- (void)nl_subscribeNext:(void (^)(id x))nextBlock completed:(void (^)(void))completedBlock {
  [self nl_subscribeNext:nextBlock error:nil completed:completedBlock];
}
- (void)nl_subscribeError:(void (^)(NSError *error))errorBlock {
  [self nl_subscribeNext:nil error:errorBlock completed:nil];
}
- (void)nl_subscribeCompleted:(void (^)(void))completedBlock {
  [self nl_subscribeNext:nil error:nil completed:completedBlock];
}
- (void)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock {
  [self nl_subscribeNext:nextBlock error:errorBlock completed:nil];
}
- (void)nl_subscribeError:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
  [self nl_subscribeNext:nil error:errorBlock completed:completedBlock];
}
- (void)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
  NLSubscriber *subscriber = [NLSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
  [self subscribe:subscriber];
}
@end

在這個類別中，將信號的 next:、error: 和 completed 以及這三個事件的組合都以 block 的形式封裝起來，從以上代碼中可以看出，這些方法最終調用的還是 - (void)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock; 方法，而它則封裝了訂閱者 NLSubsciber。

通過這麼個小小的封裝，客戶使用起來就極其方便了：

/**
   *  @brief  創建一個自定義的信號。
   *        這個信號在被訂閱時，會發送一個當前的日期值；
   *        再過三秒後，再次發送此時的日期值；
   *        最後，再發送完成事件。
   */
  RACSignal *signalInterval = [RACSignal createSignal:^(id subscriber) {
    [subscriber sendNext:[NSDate date]];
    dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(3.0 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
      [subscriber sendNext:[NSDate date]];
      [subscriber sendCompleted];
    });
    return (id)nil;
  }];
  [signalInterval nl_subscribeNext:^(id x) {
    NSLog(@"next:%@", x);
  } error:^(NSError *error) {
    NSLog(@"error:%@", error);
  } completed:^{
    NSLog(@"completed");
  }];

輸出如下：

2015-08-16 23:29:44.406 RACPraiseDemo[653:32675] next:2015-08-16 15:29:44 +0000
2015-08-16 23:29:47.701 RACPraiseDemo[653:32675] next:2015-08-16 15:29:47 +0000
2015-08-16 23:29:47.701 RACPraiseDemo[653:32675] completed

本例並沒有采用之前的 “定時器信號”，而是自己創建的信號，當有訂閱者到來時，由這個信號來決定在什麼時候發送什麼事件。這個例子裡發送的事件的邏輯請看代碼裡的注釋。

看到這裡，是不是很熟悉了？有沒有想起 subscribeNext:，好吧，我就是在使用好多好多次它之後才慢慢入門的，誰讓 RAC 的大部分教程裡面第一個講的就是它呢！

到了這裡，是不是訂閱者這部分就完了呢？我相信你也注意到了，這裡有幾個不對勁的地方：

a.無法隨時中斷訂閱操作。想想訂閱了一個無限次的定時器信號，無法中斷訂閱操作的話，定時器就是永不停止的發下去。

b.訂閱完成或錯誤時，沒有統一的地方做清理、掃尾等工作。比如現在有一個上傳文件的信號，當上傳完成或上傳錯誤時，你得斷開與文件服務器的網絡連接，還得清空內存裡的文件數據。

4、Disposable

RACDisposable

針對上述兩個問題，RACDisposable 應運而生。也就是說 Disposable 有兩個作用：

a.中斷訂閱某信號

b.訂閱完成後，執行一些收尾任務（清理、回收等等）。

訂閱者與 Disposable 的關系：

a.當 Disposable 有“清理”過，那麼訂閱者就不會再接收到這個被“清理”訂閱源的任何事件。舉例而言，就是訂閱者 subscriberX 訂閱了信號 signalA 和 signalB 兩個信號，其所對應的 Disposable 分別為 disposableA 和 disposableB，也就是說 subscriberX 會同時接收來自 signalA 和 signalB 的信號。當我們手動強制 “清理” disposableA 後，subscriberX 就不會再接收來自 signalA 的任何事件；而來自 signalB 的事件則不受影響。

b.當訂閱者 subscriberX 有接收來自任何一個信號的 “error” 或 “completed” 事件時，則不會再接收任何事件了。

可以這麼說：Disposable 代表發生了訂閱行為

根據 Disposable 的作用和與訂閱者的關系，來總結它所需要提供的接口：

a.包含清理任務的 block ;

b.執行清理任務的方法：- (void)dispose ;

c.一個用來表明是否已經 “清理” 過的布爾變量：BOOL disposed 。

咱們為這個 Disposable 也整了一個類，如下：

// .h file
/**
 *  @brief  一個 disposable 封裝了用於拆除、清理訂閱的任務工作
 */
@interface NLDisposable : NSObject
/**
 *  @brief  這個 disposable 是否已經拆除過
 */
@property (atomic, assign, getter = isDisposed, readonly) BOOL disposed;
+ (instancetype)disposableWithBlock:(void (^)(void))block;
/**
 *  @brief  執行拆除工作。能多次調用這個消息，只有第一次調用有效。
 */
- (void)dispose;
@end
// .m file
@interface NLDisposable () {
  /**
   *  @brief  類型類似於：@property (copy) void (^disposeBlock)(void);
   *       在 disposal 要執行的任務邏輯。
   *       1、如果沒有初媽值的話，則初始值默認為 `self`
   *       2、當已經 disposal 過後，值為 NULL。
   */
  void * volatile _disposeBlock;
}
@end
@implementation NLDisposable
#pragma mark Properties
- (BOOL)isDisposed {
  return _disposeBlock == NULL;
}
#pragma mark Lifecycle
- (id)init {
  self = [super init];
  if (self == nil) return nil;
  _disposeBlock = (__bridge void *)self;
  OSMemoryBarrier();
  return self;
}
- (id)initWithBlock:(void (^)(void))block {
  NSCParameterAssert(block != nil);
  self = [super init];
  if (self == nil) return nil;
  _disposeBlock = (void *)CFBridgingRetain([block copy]);
  OSMemoryBarrier();
  return self;
}
+ (instancetype)disposableWithBlock:(void (^)(void))block {
  return [[self alloc] initWithBlock:block];
}
- (void)dealloc {
  if (_disposeBlock == NULL || _disposeBlock == (__bridge void *)self) return;
  CFRelease(_disposeBlock);
  _disposeBlock = NULL;
}
#pragma mark Disposal
- (void)dispose {
  /**
   *  @brief  這裡為了讓邏輯更清晰，去掉了原子性操作。具體代碼請看 RACDisposable
   */
  if (_disposeBlock == NULL || _disposeBlock == (__bridge void *)self) return;
  void (^disposeBlock)(void) = CFBridgingRelease((void *)_disposeBlock);
  _disposeBlock = NULL;
  disposeBlock();
}
@end

從這個類提供的接口來看，顯然是做不到 “訂閱者與 Disposable 的關系” 中的第2條的。因為這條中所描述的是一個訂閱者訂閱多個信號，且能手動中斷訂閱其中一個信號的功能，而 NLDisposable 是單個訂閱關系所設計的。

RACCompoundDisposable

那怎麼組織這“多個”的關系呢？數組？Good，就是數組。OK，咱們來相像一下這個方案的初步代碼。每個訂閱者有一個 Disposable 數組，訂閱一個一個信號，則加入一個 Disposable；當手動拆除一個訂閱關系時，找到與之相關的 Disposable，發送 dispose 消息，將其從數組中移除；當訂閱者不能再接收消息時（接收過 error 或 completed 消息），要 dispose 數組中所有元素，接下來再加入元素時，直接給這個要加入的元素發送 dispose 消息；在多線程環境下，每一次加入或移除或其遍歷時，都得加鎖...（好吧，我編不下去了）

這麼復雜，看來直接用數組來維護是不可行的了。有啥其它可行的法子沒？還好，GoF 對此有個方案，叫做“組合模式”：

組合模式允許你將對象組合成樹形結構來表現 “整體/部分” 層次結構。組合能讓客戶以一致的方式處理個別對象以及對象組合。

使用組合結構，我們能把相同的操作應用在組合和個別對象上。換句話說，在大多數情況下，我們可以 忽略 對象組合和個別對象之間的差別。

本文畢竟不是來講模式的，關於這個模式更多的信息，請自行 google。

RAC 中這個組合類叫 RACCompoundDisposable， 咱們的叫 NLCompoundDisposable，來看看咱們這個類的代碼：

// .h file
#import "NLDisposable.h"
/**
 *  @brief  A disposable of disposables。當它 dispose 時，它會 dispose 
 *       它所包含的所有的 disposables。
 *
 *       如果本 compound disposable 已經 dispose 過後，再來調用 -addDisposable:，
 *     那麼其參數 disposable 會立馬調用 dispose 方法。
 * 
 *       本類中的方法說明請查看 RACCompoundDisposable 中的同名方法。
 *       本類與真正的類 RACCompoundDisposable 代碼差別較大，但本質是一樣的
 */
@interface NLCompoundDisposable : NLDisposable
/**
 *  @brief  創建並返回一個新的 compound disposable。
 */
+ (instancetype)compoundDisposable;
/**
 *  @brief  創建並返回一個新的包含了 disposables 的 compound disposable。
 *
 *  @param disposables disposable 數組
 *
 *  @return 一個新的 compound disposable
 */
+ (instancetype)compoundDisposableWithDisposables:(NSArray *)disposables;
/**
 *  @brief  將 disposable 加到本 compound disposable 中。如果本 compound disposable
 *       已經 dispose 的話，那麼參數 disposable 會被立即 dispose。
 *
 *          本方法是線程安全的。
 *
 *  @param disposable 要被加入的 disposable。如果它為 nil 的話，那麼什麼也不會發生。
 */
- (void)addDisposable:(NLDisposable *)disposable;
/**
 *  @brief  從本 compound disposable 移除指定的 disposable（不管這個 disposable 是什麼
 *       狀態）；如果這個 disposable 不在本 compound disposable 中，則什麼也不會發生。
 *
 *          本方法是線程安全的。
 *
 *  @param disposable 要被移除的 disposable。可以為 nil。
 */
- (void)removeDisposable:(NLDisposable *)disposable;
@end
// .m file
#import "NLCompoundDisposable.h"
#import @interface NLCompoundDisposable () {
  /**
   *  @brief  同步鎖
   */
  OSSpinLock _spinLock;
  /**
   *  @brief  本 compound disposable 所包含的 disposables。
   * 
   *          在操作這個數組時，應該使用 _spinLock 進行同步。如果
   *       `_disposed` 為 YES，則這個數組可能為 nil。
   */
  NSMutableArray *_disposables;
  /**
   *  @brief  本 compound disposable 是否已經 dispose 。
   *
   *          在操作這個變量時，應該使用 _spinLock 進行同步。
   */
  BOOL _disposed;
}
@end
@implementation NLCompoundDisposable
#pragma mark Properties
- (BOOL)isDisposed {
  OSSpinLockLock(&_spinLock);
  BOOL disposed = _disposed;
  OSSpinLockUnlock(&_spinLock);
  return disposed;
}
#pragma mark Lifecycle
+ (instancetype)compoundDisposable {
  return [[self alloc] initWithDisposables:nil];
}
+ (instancetype)compoundDisposableWithDisposables:(NSArray *)disposables {
  return [[self alloc] initWithDisposables:disposables];
}
- (id)initWithDisposables:(NSArray *)otherDisposables {
  self = [self init];
  if (self == nil) return nil;
  if ([otherDisposables count]) {
    _disposables = [NSMutableArray arrayWithArray:otherDisposables];
  }
  return self;
}
- (id)initWithBlock:(void (^)(void))block {
  NLDisposable *disposable = [NLDisposable disposableWithBlock:block];
  return [self initWithDisposables:@[ disposable ]];
}
- (void)dealloc {
  _disposables = nil;
}
#pragma mark Addition and Removal
- (void)addDisposable:(NLDisposable *)disposable {
  NSCParameterAssert(disposable != self);
  if (disposable == nil || disposable.disposed) return;
  BOOL shouldDispose = NO;
  OSSpinLockLock(&_spinLock);
  {
    if (_disposed) {
      shouldDispose = YES;
    } else {
      if (_disposables == nil) {
        _disposables = [NSMutableArray array];
      }
      [_disposables addObject:disposable];
    }
  }
  OSSpinLockUnlock(&_spinLock);
  if (shouldDispose) {
    [disposable dispose];
  }
}
- (void)removeDisposable:(NLDisposable *)disposable {
  if (disposable == nil) return;
  OSSpinLockLock(&_spinLock);
  {
    if (!_disposed) {
      if (_disposables != nil) {
        [_disposables removeObject:disposable];
      }
    }
  }
  OSSpinLockUnlock(&_spinLock);
}
#pragma mark RACDisposable
- (void)dispose {
  NSArray *remainingDisposables = nil;
  OSSpinLockLock(&_spinLock);
  {
        _disposed = YES;
        remainingDisposables = _disposables;
        _disposables = nil;
  }
  OSSpinLockUnlock(&_spinLock);
  if (remainingDisposables == nil) return;
  [remainingDisposables makeObjectsPerformSelector:@selector(dispose)];
}
@end

RACScheduler 簡介

本文不打算研究 RACScheduler 源碼，但其又是 RAC 中不可或缺的一個組件，在研究 RACSignal 的源碼時不可避免地會遇到它，所以對其作下介紹還是有必要的。其實它的源碼並不復雜，可自行研究。

ReactiveCocoa 中 RACSignal 發送的所有事件的傳遞交給了一個特殊的框架組件——調度器，即 RACScheduler 類簇（類簇模式稍後介紹）。調度器是為了簡化 同步/異步/延遲 事件傳遞 以及 取消預定的任務(scheduded actions) 這兩種 RAC 中常見的動作而提出來的。“事件傳遞” 簡單而言就是些 blocks，RACScheduler 所做的就是：調度這些 blocks (schedule blokcs，還是英文的意思准確些)。我們可以通過那些調度方法所返回的 RACDisposable 對象來取消那些 scheduling blocks。

正如前面所說，RACScheduler 是一個類簇。咱們來看看幾種具體的調度器：

a.RACImmediateScheduler

這是 RAC 內部使用的私有調度器，只支持同步 scheduling。就是簡單的馬上執行 block。這個調試器的延遲 scheduling 是通過調用 -[NSThread sleepUntilDate:] 來阻塞當前線程來達到目的的。顯然，這樣一個調度器，沒法取消 scheduling，所以它那些方法返回的 disposables 啥也不會做（實際上，它那些 scheduling 方法返回的是nil）。

b.RACQueueScheduler

這個調度器使用 GCD 隊列來 scheduling blocks。如果你對 GCD 有所了解的話，你會發現這個調度器的功能很簡單，它只是在 GCD 隊列 dispatching blocks 上的簡單封裝罷了。

c.RACSubscriptionScheduler

這是另一個內部使用的私有調度器。如果當前線程有調度器（調度器可以與線程相關聯起來：associated）那它就將 scheduling 轉發給這個線程的調度器；否則就轉發給默認的 background queue 調試器。

接口

調試器有下面一些方法：

- (RACDisposable *)schedule:(void (^)(void))block;

- (RACDisposable *)after:(NSDate *)date schedule:(void (^)(void))block;

- (RACDisposable *)afterDelay:(NSTimeInterval)delay schedule:(void (^)(void))block;

- (RACDisposable *)after:(NSDate *)date repeatingEvery:(NSTimeInterval)interval withLeeway:(NSTimeInterval)leeway schedule:(void (^)(void))block;

scheduling block 如下：

RACDisposable *disposable = [[RACScheduler mainThreadScheduler] afterDelay:5.0 schedule:^{
// do something
}];

// 如果你想要取消 scheduling block
[disposable dispose]; // block scheduling 被取消了，不會再被執行。

5、Subscriber 和 Disposable

前面介紹了 Disposable 的來源，現在來研究下怎麼使用它。還記得嗎，訂閱者與信號打交道的唯一方式是 RACSignal 中的一個方法：

- (RACDisposable *)subscribe:(id
)subscriber;

自定義信號所對應的類是 RACDynamicSignal。RACSignal 采用的是類簇模式。除自定義信號之外還有幾種其它的信號，之後會研究到。OC 中的 NSNumber 用的就是類簇模式。類簇是Foundation框架中廣泛使用的設計模式。類簇將一些私有的、具體的子類組合在一個公共的、抽象的超類下面，以這種方法來組織類可以簡化一個面向對象框架的公開架構，而又不減少功能的豐富性。

咱們來研究一下自定義信號裡的這個方法的實現。這個方法實現的難處在於：“一個訂閱者可以訂閱多個信號，並可以手動拆除其中任何一個訂閱”。針對這個問題，提出了上節講到的　RACDisposable。也就是說，在每一次訂閱時，都會返回一個與這次訂閱相關的 Disposable，那怎麼做到這一點呢？

給訂閱者添加一個 CompoundDisposable 類型的屬性 （畢竟 CompoundDisposable 就是用來針對多個 Disposable 的統一管理而存在的），然後在每一次訂閱時，都加一個 Disposable 到這個屬性裡，行不行？但很可惜，訂閱者是一個協議 protocol RACSubscriber，而不是一個具體的類，咱們在使用到它時，都是別人實現了這個協議的類的對象，所以咱們不太可能做到說給這麼一個未知的類添加一個屬性。

事實上，RAC 中確實有 RACSubscriber 這麼一個私有類（它是咱們第一個自定義類 NLSubscriber 的原型），咱們叫它做 class RACSubscriber。嗯，class RACSusbscriber 實現了 protocol RACSubscriber 協議：@interface RACSubscriber : NSObject。有沒有想到 class NSObject 和 protocol NSObject ？雖然它們形式上確實很像，但千萬別混為一談。RAC 中的其它實現了 protocol RACSubscriber 協議的訂閱者類可沒有一個繼承自 class RACSubscriber 的。

咱們可以用裝飾模式來解決這個問題

裝飾模式。在不必改變原類文件和使用繼承的情況下，動態地擴展一個對象的功能。

訂閱者裝飾器 RACPassthroughSubscriber

在訂閱者每一次訂閱信號時產生一個 Disposable，並將其與此次訂閱關聯起來，這是通過裝飾器 RACPassthroughSubscriber 來做到的。這個裝飾器的功能：

a.包裝真正的訂閱者，使自己成為訂閱者的替代者。

b.將真正的訂閱者與一個訂閱時產生的 Disposable 關聯起來。

這正是一個裝飾器所應該做的。依之前的，咱們來模仿這個裝飾器，新建一個咱們的裝飾器：NLPassthroughSubscriber，來看下它的代碼：

// .f file
@class RACCompoundDisposable;
@class RACSignal;
/**

*  @brief  這是一個訂閱者的裝飾器。在沒有 dispose 時，它會把接收到的所有
*       的事件都轉發給真實的訂閱者。
*/
@interface NLPassthroughSubscriber : NSObject
/**

*  @brief  初始化方法
*
*  @param subscriber 被包裝的真實的訂閱者，本裝飾器會把接收到的所有的事件都轉發給這個訂閱者。不能為 nil
*  @param signal     要發送事件給這個裝飾器的信號。
*  @param disposable 當這個 disposable 接收到 dispose 消息後，將不會再轉發事件。不能為 nil
*
*  @return 返回一個初始化後的 passthrough subscriber
*/
- (instancetype)initWithSubscriber:(id)subscriber signal:(RACSignal *)signal disposable:(RACCompoundDisposable *)disposable;

@end

// .m file
@interface NLPassthroughSubscriber ()

// 被轉發事件的訂閱者。
@property (nonatomic, strong, readonly) idinnerSubscriber;

// 要發送事件給本裝飾器的信號
@property (nonatomic, unsafe_unretained, readonly) RACSignal *signal;

// 代表當前訂閱關系的 disposable。當它 dispose 後，將不會再轉發任何事件給 `innerSubscriber`。
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;

@end

@implementation NLPassthroughSubscriber

#pragma mark Lifecycle

- (instancetype)initWithSubscriber:(id)subscriber signal:(RACSignal *)signal disposable:(RACCompoundDisposable *)disposable {
NSCParameterAssert(subscriber != nil);

self = [super init];
if (self == nil) return nil;

_innerSubscriber = subscriber;
_signal = signal;
_disposable = disposable;

/**
*  告訴訂閱者：發生了訂閱行為。並將這次訂閱行為相關的 `Disposable` 傳給訂閱者。
*/
[self.innerSubscriber didSubscribeWithDisposable:self.disposable];
return self;
}
#pragma mark RACSubscriber
- (void)sendNext:(id)value {
/**
*  如果 disposable 已經 dispose 過，就不再轉發事件
*/
if (self.disposable.disposed) return;

/**
*  轉發 next 事件
*/
[self.innerSubscriber sendNext:value];
}

- (void)sendError:(NSError *)error {
if (self.disposable.disposed) return;

[self.innerSubscriber sendError:error];
}

- (void)sendCompleted {
if (self.disposable.disposed) return;

[self.innerSubscriber sendCompleted];
}

- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable {
if (disposable != self.disposable) {
[self.disposable addDisposable:disposable];
}
}

@end

自定義信號 RACDynamicSignal 的訂閱方法 subscribe

咱們來看看 RACDynamicSignal 是怎麼來使用 RACPassthroughSubscriber 的，這裡就不自己寫代碼了，直接上它的代碼：

- (RACDisposable *)subscribe:(id)subscriber {
NSCParameterAssert(subscriber != nil);

/**
*  本次訂閱相關 disposable。本方法的返回值，起 拆除 本次訂閱的作用。
*/
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];

/**
* 訂閱者裝飾器。
*/
subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

/**
*  _didSubscriber 是在 `+ createSignal` 方法中進入的 block 參數。
*  + (RACSignal *)createSignal:(RACDisposable * (^)(idsubscriber))didSubscribe;
*  這個 block 以 subscriber 為參數，返回一個 disposable，即 innerDisposable，而這個 innerDisposable
*  的作用是在 subscriber 不再訂閱本 signal 時，起回收資源的作用。
*/
if (self.didSubscribe != NULL) {
RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
RACDisposable *innerDisposable = self.didSubscribe(subscriber);
[disposable addDisposable:innerDisposable];
}];

[disposable addDisposable:schedulingDisposable];
}

return disposable;
}

可以看到，訂閱者裝飾器直接偽裝成真正的訂閱器，傳給 didSubscribe 這個 block 使用。在這個 block 中，會有一些事件發送給訂閱者裝飾器，而這個訂閱者裝飾器則根據 disposable 的狀態來來決定是否轉發給真正的訂閱者。disposable 作為返回值，返回給外部，也就是說能夠從外部來取消這個訂閱了。

RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
RACDisposable *innerDisposable = self.didSubscribe(subscriber);
[disposable addDisposable:innerDisposable];
}];

[disposable addDisposable:schedulingDisposable];

從這幾行代碼中，我們可以看到，didSubscribe 這個 block 是處於 subscriptionScheduler 這個 scheduler 的調度中。RACSubscriptionScheduler 的調度是取決於當前所在的線程的，即 didSubscribe 可能會在不同的調度器中被執行。

假設當前 -(RACDisposable *)subscribe:(id)subscriber 這個方法是在異步環境下調用的，那麼在 disposable 返回後，在schedule block 還沒有來得及調用，此時 disposable 中包含 schedulingDisposable。如果我們此時給 disposable 發送 dispose 消息，那麼 schedulingDisposable 也會被 dispose，schedule block 就不會執行了；如果是在 schedule block 執行中或執行後給 disposable 發送 dispose 消息，那麼 innerDisposable 和 schedulingDisposable 都會被 dispose。這些行為正是咱們所預期的。

6、再次改進NLSubscriber

a.didSubscribeWithDisposable

@protocol RACSubscriber
@required
- (void)didSubscribeWithDisposable:(RACCompoundDisposable *)disposable;
@end

這個 RACSubscriber 協議中聲明的一個方法，在最開始的時候被我們特意給忽略，現在是時候回過頭來看看它了。對於一個訂閱者來說，next、error 和 completed 三種事件分別對應協議裡的三種方法，那麼這個方法存在的意義是什麼呢？

從 RACSubscriber 協議中，可以看到，當一個訂閱者有收到過 error 或 completed 事件後，這個訂閱者就不能再接收任何事件了，換句話說，此時這個訂閱者會解除所有的訂閱關系，且無法再次訂閱。既然要解除所有訂閱，首先我得知道我訂閱過哪些信號是不？而代表一個訂閱行為的就是 disposable ，告訴它就傳一個給它好了。所以這個方法就是告訴訂閱者：你發生了訂閱行為。

那為啥要 RACCompoundDisposable 類型作為參數呢？因為有些訂閱者會針對其附加一些操作，而只有這個類型的 disposable 才能動態加入一些操作。接下來我們就會看到的。

b.NLSubscriber 結合 RACDisposable

這一次改進 NLSubscriber 的目的是讓其可以終結自己的訂閱能力的功能。同時實現 didSubscribeWithDisposable 方法。千言萬語不如實際代碼，讓我們來一探究竟：

#import "NLSubscriber.h"
#import @interface NLSubscriber ()
@property (nonatomic, copy) void (^next)(id value);
@property (nonatomic, copy) void (^error)(NSError *error);
@property (nonatomic, copy) void (^completed)(void);
/**
 *  @brief  代表訂閱者本身總體訂閱行為的 disposable。
 *       當有接收到 error 或 completed 事件時，應該 dispose 這個 disposable。
 */
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;
@end
@implementation NLSubscriber
#pragma mark Lifecycle
+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
  NLSubscriber *subscriber = [[self alloc] init];
  subscriber->_next = [next copy];
  subscriber->_error = [error copy];
  subscriber->_completed = [completed copy];
  return subscriber;
}
- (instancetype)init {
  self = [super init];
  if (self == nil) return nil;
  @unsafeify(self);
  /**
   *  當 _disposable 被發送 dispose 消息時，將 next、error 和 completed 這三個
   *  block 設置為 nil，從而間實現訂閱者無法再接收任何事件的功能。
   */
  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;
  /**
   *  將 otherDisposable 添加到 selfDisposable 中。這樣當 selfDisposable 被 dispose 時，
   *  otherDisposable 也能被 dispose。
   * 
   *  這樣，當訂閱者接收到 error 或 completed 事件時，就能解除這個訂閱者自身的所有訂閱行為了。
   */
  RACCompoundDisposable *selfDisposable = self.disposable;
  [selfDisposable addDisposable:otherDisposable];
  @unsafeify(otherDisposable);
  /**
   *  如果這個訂閱行為被解除，就將 otherDisposable 從 selfDisposable 中移除。
   *  (我們給 otherDisposable 增加了行為，這也就是參數需要是 RACCompoundDisposable
   *  類型的原因了。當然，其它的訂閱者怎麼用這個參數就跟其實際的業務相關了。)
   */
  [otherDisposable addDisposable:[RACDisposable disposableWithBlock:^{
    @strongify(otherDisposable);
    [selfDisposable removeDisposable:otherDisposable];
  }]];
}
@end

c.改進類別 nl_Subscription

還記得麼？nl_Subscription 類別中的訂閱方法一旦訂閱，就無法停止了，這顯然有很大的問題。解決這個問題很簡單，直接將 disposable 返回即可：

// .h file
@interface RACSignal (nl_Subscription)
...
- (RACDisposable *)nl_subscribeError:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock;

@end

// .m file
@implementation RACSignal (nl_Subscription)
...
- (RACDisposable *)nl_subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
  NLSubscriber *subscriber = [NLSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
  return [self subscribe:subscriber];
}

RACSignal - Operations

本節主要研究這些操作（Operations） —— flattenMap:、map:、 filter: ….

終於看到你想看的東西了？好吧，我承認，上節的東西很無趣，可能壓根不是你想看的東西。但如果沒弄清上面的內容的話，直接研究 Operations 可是會比較吃力的喲~

你以為咱們現在開始研究 Operations？哈哈，你又得失望了~ 咱得先看看這兩個類：RACEmptySignal 和 RACReturnSignal。

1、兩個 RACSignal 的特殊子類 RACEmptySignal 和 RACReturnSignal

a.RACEmptySignal

RACEmptySignal 是 +[RACSignal empty] 的內部實現，一個私有 RACSignal 子類。它就是一個會立即 completed 的信號。讓我們來看看它的 - subscribe: 方法：

- (RACDisposable *)subscribe:(id)subscriber {
    NSCParameterAssert(subscriber != nil);
    
    /**
     *  只要一訂閱，就給 subscriber 發送 completed 事件。
     */
    return [RACScheduler.subscriptionScheduler schedule:^{
        [subscriber sendCompleted];
    }];
}

這樣一個訂閱者一訂閱就會 completed 信號有什麼用呢？稍後揭曉。

b.RACReturnSignal

RACReturnSignal 是 +[RACSignal return:] 的內部實現，也是一個私有 RACSignal 子類。它會同步發送出一個值（即 next）給訂閱者，然後再發送 completed 事件。 它比 RACEmptySignal 多了一點點東西，它是。直接看其實現：

@interface RACReturnSignal ()
// 在本信號被訂閱時會發送的值。
@property (nonatomic, strong, readonly) id value;
@end

@implementation RACReturnSignal

#pragma mark Lifecycle
+ (RACSignal *)return:(id)value {
    RACReturnSignal *signal = [[self alloc] init];
    signal->_value = value;
    return signal;
}

#pragma mark Subscription

- (RACDisposable *)subscribe:(id)subscriber {
    NSCParameterAssert(subscriber != nil);
   
    return [RACScheduler.subscriptionScheduler schedule:^{
        [subscriber sendNext:self.value];
        [subscriber sendCompleted];
    }];
}

@end

純吐槽：為啥要叫 ReturnSignal 呢？不如直接 OneValueSignal 好了。O(∩_∩)O~~ 不過說真的，RAC 的命名真心不咋地。

那麼發送一個 next 後又 completed 的信號又有啥用呢？等下會知道地。

b.concat: 練手

-[RACSignal concat:] 是源碼較簡單，且使用頻率也較多的。那咱們就來拿它來練練手好了。

- (RACSignal *)concat:(RACSignal *)signal {
    return [[RACSignal createSignal:^(id subscriber) {
        RACSerialDisposable *serialDisposable = [[RACSerialDisposable alloc] init];
       
        RACDisposable *sourceDisposable = [self subscribeNext:^(id x) {
            [subscriber sendNext:x];
        } error:^(NSError *error) {
            [subscriber sendError:error];
        } completed:^{
            RACDisposable *concattedDisposable = [signal subscribe:subscriber];
            serialDisposable.disposable = concattedDisposable;
        }];
        
        serialDisposable.disposable = sourceDisposable;
        return serialDisposable;
    }] setNameWithFormat:@"[%@] -concat: %@", self.name, signal];
}

RACSerialDisposable 是 RACDisposable 的子類，它包含一個 Disposable，能夠在運行時設置這個 Disposable。當設置新的 newDisposable時，老的 oldDisposable 會被 dispose。當 RACSerialDisposable 被 dispose 時，其所包含的 Disposable 會被 dispose。

基本上，對一個 RACSignal 的操作的返回值是一個新的 RACSignal 值時，其內部都是調用了 +[RACSignal createSignal:] 這個方法。這個創建信號返回的實際是自定義信號：RACDynamicSignal，針對它前文有所介紹。

這裡有一個小技巧。因為很多信號的操作是針對該信號本身 self 所發送的值作的操作。那也就是說會訂閱 self，那咱們先找到這一句再說：self subscribe: 或 self subscribeNext:...。嗯，找到了這幾行：

RACDisposable *sourceDisposable = [self subscribeNext:^(id x) {
    [subscriber sendNext:x];
} error:^(NSError *error) {
    [subscriber sendError:error];
} completed:^{
    RACDisposable *concattedDisposable = [signal subscribe:subscriber];
    serialDisposable.disposable = concattedDisposable;
}];

在訂閱了 self 後，將 next 和 error 事件發送給訂閱者 subscriber。當 self 發送了 completed 事件事，再讓 subscriber 訂閱參數 signal。也就是當源信號完成後訂閱 signal。怎麼樣，很簡單吧。

c.zipWith:

再來一個練手的玩意。-[RACSignal zipWith:] 比 -[RACSignal concat:] 稍微復雜點。它是將 self 和 參數 signal 兩個信號發送的值合並起來發送給訂閱者。

- (RACSignal *)zipWith:(RACSignal *)signal {
    NSCParameterAssert(signal != nil);
    return [[RACSignal createSignal:^(id subscriber) {
        __block BOOL selfCompleted = NO;
        NSMutableArray *selfValues = [NSMutableArray array];
        __block BOOL otherCompleted = NO;
        NSMutableArray *otherValues = [NSMutableArray array];
        void (^sendCompletedIfNecessary)(void) = ^{
            @synchronized (selfValues) {
                BOOL selfEmpty = (selfCompleted && selfValues.count == 0);
                BOOL otherEmpty = (otherCompleted && otherValues.count == 0);
                if (selfEmpty || otherEmpty) [subscriber sendCompleted];
            }
        };
        void (^sendNext)(void) = ^{
            @synchronized (selfValues) {
                if (selfValues.count == 0) return;
                if (otherValues.count == 0) return;
                RACTuple *tuple = [RACTuple tupleWithObjects:selfValues[0], otherValues[0], nil];
                [selfValues removeObjectAtIndex:0];
                [otherValues removeObjectAtIndex:0];
                [subscriber sendNext:tuple];
                sendCompletedIfNecessary();
            }
        };
        RACDisposable *otherDisposable = [signal subscribeNext:^(id x) {
            @synchronized (selfValues) {
                [otherValues addObject:x ?: RACTupleNil.tupleNil];
                sendNext();
            }
        } error:^(NSError *error) {
            [subscriber sendError:error];
        } completed:^{
            @synchronized (selfValues) {
                otherCompleted = YES;
                sendCompletedIfNecessary();
            }
        }];
        return [RACDisposable disposableWithBlock:^{
            [selfDisposable dispose];
            [otherDisposable dispose];
        }];
    }] setNameWithFormat:@"[%@] -zipWith: %@", self.name, signal];
}

同樣的，重點在 [self subscriberNext:] 和 [signal subscribeNext:] 處。這裡的實現是訂閱 self 和 signal 信號，然後將它們發送出的值收集起來，當兩個都發出了值時，分別拿出兩個信號最早發出的值，合並為一個 RACTuple，再發送給訂閱者 subscriber。這個也很簡單吧，只是代碼稍多點而已。

d.bind:

（i）說明

信號的很多 operations 的實現調用來調用去最後都是調用了這個 -[RACSignal bind:] 方法，比如 flattenMap: 、map:、filter 等等。那咱們就來看看這個方法是哪路神仙？

這是在 RACStream 中聲明的抽象方法。來看看它的聲明：

typedef RACStream * (^RACStreamBindBlock)(id value, BOOL *stop);
- (instancetype)bind:(RACStreamBindBlock (^)(void))block;

RACStreamBindBlock 是一個 block。它從一個 RACStream 中接收一個值，並且返回一個與該流相同類型的實例。如果將 stop 設為 YES，則會在返回一個實例後終結此次 bind。如果返回 nil 則會立即終結。

bind: 方法是將流中每一個值都放到 RACStreamBindBlock 中跑一下。來看看其參數：block。然而這有什麼卵用呢？好吧，我太笨，從它的說明來看，我真的不能理解它有什麼用。

（ii）源碼解讀

既然從方法說明了解不到，那直接來看其源碼了。

- (RACSignal *)bind:(RACStreamBindBlock (^)(void))block {
    NSCParameterAssert(block != NULL);
    return [[RACSignal createSignal:^(id subscriber) {
        RACStreamBindBlock bindingBlock = block();
        NSMutableArray *signals = [NSMutableArray arrayWithObject:self];
        RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
        // 三.
        void (^completeSignal)(RACSignal *, RACDisposable *) = ^(RACSignal *signal, RACDisposable *finishedDisposable) {
            BOOL removeDisposable = NO;
            @synchronized (signals) {
                [signals removeObject:signal];
                if (signals.count == 0) {
                    [subscriber sendCompleted];
                    [compoundDisposable dispose];
                } else {
                    removeDisposable = YES;
                }
            }
            if (removeDisposable) [compoundDisposable removeDisposable:finishedDisposable];
        };
        // 二.
        void (^addSignal)(RACSignal *) = ^(RACSignal *signal) {
            @synchronized (signals) {
                [signals addObject:signal];
            }
            RACSerialDisposable *selfDisposable = [[RACSerialDisposable alloc] init];
            [compoundDisposable addDisposable:selfDisposable];
            RACDisposable *disposable = [signal subscribeNext:^(id x) {
                [subscriber sendNext:x];
            } error:^(NSError *error) {
                [compoundDisposable dispose];
                [subscriber sendError:error];
            } completed:^{
                @autoreleasepool {
                    completeSignal(signal, selfDisposable);
                }
            }];
            selfDisposable.disposable = disposable;
        };
        @autoreleasepool {
            RACSerialDisposable *selfDisposable = [[RACSerialDisposable alloc] init];
            [compoundDisposable addDisposable:selfDisposable];
           //  一.
            RACDisposable *bindingDisposable = [self subscribeNext:^(id x) {
                // Manually check disposal to handle synchronous errors.
                if (compoundDisposable.disposed) return;
                BOOL stop = NO;
                id signal = bindingBlock(x, &stop);
                @autoreleasepool {
                    if (signal != nil) addSignal(signal);
                    if (signal == nil || stop) {
                        [selfDisposable dispose];
                        completeSignal(self, selfDisposable);
                    }
                }
            } error:^(NSError *error) {
                [compoundDisposable dispose];
                [subscriber sendError:error];
            } completed:^{
                @autoreleasepool {
                    completeSignal(self, selfDisposable);
                }
            }];
            selfDisposable.disposable = bindingDisposable;
        }
        return compoundDisposable;
    }] setNameWithFormat:@"[%@] -bind:", self.name];
}

我們一步一步來看。先從第 一 步開始，其步驟如下：

• 訂閱 self

• 針對 self 發出的每一個值 x，經過 bindingBlock，獲取一個信號：signal

    1. 如果 signal 不為 nil，就轉到第二步：addSignal

    2. 如果 signal 為 nil，或 stop 為 YES，則轉到第三步：completedSignal

    3. 如果 self 發出 error 事件，則中斷訂閱；如果 self 發出 completed 事件則轉到第三步：completedSignal

第二步：addSignal:signal

• 先將 signal 添加到 signals中

• 訂閱 signal

    1. 將 signal 的 next 事件轉發給訂閱者 subscriber

    2. 如果 signal 發送 error 事件則中斷訂閱

    3. 如果 signal 發送 complete 事件，則轉到第三步

第三步：completeSignal:signal:disposable

• 將 signal 從 signals 中移除

• 如果 signals 中沒有了 signal，那麼訂閱就完成了

好了，來總結一下這個 -bind: ：

• 訂閱原信號 self 的 values。

• 將 self 發出的任何一個值，都對其使用 bindingBlock 進行轉換。

• 如果 bindingBlock 返回一個信號，則訂閱它，將從它那接收到的每個值都傳遞給訂閱者 subscriber。

• 如果 bindingBlock 要求結束綁定，則 complete self 信號。

• 如果 所有 的信號全都 complete，則給 subscriber 發送 completed 事件.

• 如果任何一個信號發出 error，將其發送給 subscriber。

那從中可以玩出什麼花樣呢？

（iii）示例

咱們先用用它，再看看能怎麼玩吧。

示例1：結合 RACReturnSignal

RACSignal *signalInterval = [[RACSignal interval:1.0 onScheduler:[RACScheduler mainThreadScheduler]] take:3];
  RACSignal *bindSignal = [signalInterval bind:^RACStreamBindBlock{
    return ^(id value, BOOL *stop) {
      NSLog(@"inner value: %@", value);
      return [RACSignal return:value];
    };
  }];
  [bindSignal subscribeNext:^(id x) {
    NSLog(@"outer value: %@", x);
  }];

輸出如下：

2015-08-27 17:16:17.933 RACPraiseDemo[3063:168556] inner value: 2015-08-27 09:16:17 +0000
2015-08-27 17:16:17.934 RACPraiseDemo[3063:168556] outer value: 2015-08-27 09:16:17 +0000
2015-08-27 17:16:18.931 RACPraiseDemo[3063:168556] inner value: 2015-08-27 09:16:18 +0000
2015-08-27 17:16:18.931 RACPraiseDemo[3063:168556] outer value: 2015-08-27 09:16:18 +0000
2015-08-27 17:16:19.931 RACPraiseDemo[3063:168556] inner value: 2015-08-27 09:16:19 +0000
2015-08-27 17:16:19.931 RACPraiseDemo[3063:168556] outer value: 2015-08-27 09:16:19 +0000

這個示例就是在 bind: 中簡單的返回值。那咱們將這個值變化一下如何？

示例2：結合 RACReturnSignal、轉換 value

RACSignal *signalInterval = [[RACSignal interval:1.0 onScheduler:[RACScheduler mainThreadScheduler]] take:3];
  RACSignal *bindSignal = [signalInterval bind:^RACStreamBindBlock{
    return ^(NSDate *value, BOOL *stop) {
      NSLog(@"inner value: %@", value);
      NSTimeInterval nowTime = [value timeIntervalSince1970];
      return [RACSignal return:@(nowTime)];
    };
  }];
  [bindSignal subscribeNext:^(id x) {
    NSLog(@"outer value: %@", x);
  }];

輸出如下：

2015-08-27 17:34:04.938 RACPraiseDemo[3153:176383] inner value: 2015-08-27 09:34:04 +0000
2015-08-27 17:34:04.939 RACPraiseDemo[3153:176383] outer value: 1440668044.936496
2015-08-27 17:34:05.939 RACPraiseDemo[3153:176383] inner value: 2015-08-27 09:34:05 +0000
2015-08-27 17:34:05.939 RACPraiseDemo[3153:176383] outer value: 1440668045.939163
2015-08-27 17:34:06.941 RACPraiseDemo[3153:176383] inner value: 2015-08-27 09:34:06 +0000
2015-08-27 17:34:06.941 RACPraiseDemo[3153:176383] outer value: 1440668046.941275

哇哇，這就是個 map: 有木有？ 現在，有感受到 RACReturnSignal 的魅力？RACReturnSignal 和 -bind: 結合能轉換 value。

示例3：結合 RACEmptySignal

現在來換個玩法試試看，這回換 RACEmptySignal 來玩玩。

RACSignal *signalInterval = [[RACSignal interval:1.0 onScheduler:[RACScheduler mainThreadScheduler]] take:3];
  __block NSUInteger count = 0;
  RACSignal *bindSignal = [signalInterval bind:^RACStreamBindBlock{
    return ^(NSDate *value, BOOL *stop) {
      NSLog(@"inner value: %@", value);
      ++count;
      if (count % 2 == 0) {
        return [RACSignal empty];
      }
      return [RACSignal return:value];
    };
  }];
  [bindSignal subscribeNext:^(id x) {
    NSLog(@"outer value: %@", x);
  }];

輸出如下：

2015-08-27 17:53:45.345 RACPraiseDemo[3363:188270] inner value: 2015-08-27 09:53:45 +0000
2015-08-27 17:53:45.346 RACPraiseDemo[3363:188270] outer value: 2015-08-27 09:53:45 +0000
2015-08-27 17:53:46.345 RACPraiseDemo[3363:188270] inner value: 2015-08-27 09:53:46 +0000
2015-08-27 17:53:47.342 RACPraiseDemo[3363:188270] inner value: 2015-08-27 09:53:47 +0000
2015-08-27 17:53:47.342 RACPraiseDemo[3363:188270] outer value: 2015-08-27 09:53:47 +0000

這一次，“outer value” 比 “inner value” 少了一個，這就是 filter: 呀！RACEmptySignal 與 bind: 結合能過濾 value。

示例4：改進 bind:

經過這幾個示例，我們可以發現，直接使用 bind: 是比較麻煩的。而一般情況下，咱們還真用不到 stop，那咱們就改進一下呗：

- (instancetype)flattenMap:(RACStream * (^)(id value))block {
    Class class = self.class;
    return [[self bind:^{
        return ^(id value, BOOL *stop) {
            /**
             *  如果 block 返回 nil，得用 RACEmptySignal 代替，
             *  不然會結束 `bind:`
             */
            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:了。不必過多解釋了吧~

（iv）-map:

嗯，這其實就是 -flattenMap: 與 RACReturnSignal 的結合：

- (instancetype)map:(id (^)(id value))block {
    NSCParameterAssert(block != nil);
    Class class = self.class;
    return [[self flattenMap:^(id value) {
        return [class return:block(value)];
    }] setNameWithFormat:@"[%@] -map:", self.name];
}

（v）-flatten

信號可以發送類型的值，當然也包括 RACSignal 類型。例如，RACCommand 的 executionSignals 這個信號，它發出的值就是 RACSignal 類型的。對於這種發出的值是 RACSignal 類型的 RACSignal，叫做 signal of signals。這有點類似於 disposable of disposables。

既然這個信號發出的就是 RACSignal，那在 -flattenMap:中，我們直接將 value 返回就好了。來看看示例：

/**
   *  有效三次的間隔為1秒定時器信號，此時 signalInterval 是一個 signal of NSDates
   */
  RACSignal *signalInterval = [[RACSignal interval:1 onScheduler:[RACScheduler mainThreadScheduler]] take:3];
  /**
   *  將定時器信號裡的值修改成 RACSignal 類型
   *  此時，signalInterval 變成了一個 signal of signals
   */
  signalInterval = [signalInterval map:^id(NSDate *date) {
    return [RACSignal return:date];
  }];
  /**
   *  既然 signalInterval 裡的值都是信號，那直接將這些信號返回即可
   */
  RACSignal *signal = [signalInterval flattenMap:^RACStream *(RACSignal *returnSignal) {
    return returnSignal;
  }];
  /**
   *  由於 signalInterval 裡的值都是包含了一個 NSDate 值的 RACReturnSignal,
   *  經過 `-flattenMap:` 過後，signal 就變成了 signal of NSDates。
   */
  [signal subscribeNext:^(id x) {
    NSLog(@"value: %@", x);
  }];

輸出如下：

2015-08-27 21:16:29.517 RACPraiseDemo[549:11996] value: 2015-08-27 13:16:29 +0000
2015-08-27 21:16:30.516 RACPraiseDemo[549:11996] value: 2015-08-27 13:16:30 +0000
2015-08-27 21:16:31.516 RACPraiseDemo[549:11996] value: 2015-08-27 13:16:31 +0000

（vi）小結

RACSignal 的 operations 實在太多，全部在這裡列出來不現實，也沒有這個必要。我相信，經過前面的解析，你現在再去看其它 的一個 operation 源碼，也應該不是太大的難事。

RAC() 宏展開

RAC 的最大的魅力之一就是綁定：RAC(self, ...) = signal; 這應該是大家經常寫的一條語句。有沒有想過它是怎麼工作的呢？咱們來看點代碼：

@property (nonatomic, strong) NSString *text;
//--------
RACSignal *signal = [[RACSignal interval:1 onScheduler:[RACScheduler mainThreadScheduler]] take:3];
  signal = [signal map:^id(id value) {
    return [value description];
  }];
RAC(self, text) = signal;

重點在 RAC(self, text) = signal; 這一行。先來看看將這個宏展開是什麼樣子（RAC 對宏的運用很是牛B，有興趣請看這篇文章）：

[[RACSubscriptingAssignmentTrampoline alloc] initWithTarget:(self) nilValue:nil][@"text"] = signal;

看得更清楚一點：

RACSubscriptingAssignmentTrampoline *assignment = [[RACSubscriptingAssignmentTrampoline alloc] initWithTarget:(self) nilValue:nil];
assignment[@"text"] = signal;

跳到 RACSubscriptingAssignmentTrampoline 類的聲明，可以看到：

@interface RACSubscriptingAssignmentTrampoline : NSObject
- (id)initWithTarget:(id)target nilValue:(id)nilValue;

// 這是可以使用下標 `[]` 語法的關鍵
- (void)setObject:(RACSignal *)signal forKeyedSubscript:(NSString *)keyPath;
@end

這個類使用了 clang 的特性，可以使用 []語法（[] 的相關文章）。也就是說 assignment[@"text"] = signal;，實際上是這樣子的：

[assignment setObject:signal forKeyedSubscript:@"text"];

再看 - (void)setObject:(RACSignal *)signal forKeyedSubscript:(NSString *)keyPath; 這個方法的實現，我們發現，它其實調用的是 signal 的方法：-

- (RACDisposable *)setKeyPath:(NSString *)keyPath onObject:(NSObject *)object nilValue:(id)nilValue {
    ...
    RACDisposable *subscriptionDisposable = [self subscribeNext:^(id x) {
        ...
        [object setValue:x ?: nilValue forKeyPath:keyPath];
    }...
    ...
}

哦，原來它就是訂閱了 signal，並將 signal 發出的每一值都設置給 object 的 keyPath 屬性而已。很簡單嘛~

結束

本文研究了 RAC 中的一些基本組件，並沒有對一些高級內容進行深入研究，所以才叫“淺析”。但這些也是對高級內容深入研究的基礎，既然有“漁”，何懼無“魚”呢？

其實頗想繼續分享，但心有余而力不足。

還可研究的主題：

• Subjects 它也是 RACSignal 一些操作的基礎，值得研究。難度系數：2 （最高為5）

• RACMulticastConnection 常用，值得研究。難度系數：3

• Foundation、UIKit、KVO （給各系統類加的 rac_ 擴展），有研究價值。研究過後，你會對 runtime 會有很深入的了解，還會接觸到一些 OC 中少用的知識（如 NSProxy 等），能開拓視野。難度系數：5

難度系數是本人 YY 出來的，別較真，僅當參考。