一、订阅者都持有disposable

1、 @interface RACPassthroughSubscriber ()
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;

2、 @interface RACSubject ()
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;

3、@interface RACSubscriber ()
@property (nonatomic, strong, readonly) RACCompoundDisposable *disposable;

4、RACChannel
@property (nonatomic, strong, readonly) id<RACSubscriber> otherTerminal;

RacPassthroghSubscriber
RacSubject，
RacSubscriber
Racchannel
didSubscribeWithDisposable

[self.otherTerminal didSubscribeWithDisposable:disposable];
[self.innerSubscriber didSubscribeWithDisposable:self.disposable];

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

• (void)didSubscribeWithDisposable:(RACCompoundDisposable *)d {
  if (d.disposed) return;
  [self.disposable addDisposable:d];

  @weakify(self, d);
  [d addDisposable:[RACDisposable disposableWithBlock:^{
  @strongify(self, d);
  [self.disposable removeDisposable:d];
  }]];
  }

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

innerSubscriber 具有一个多态性的概念 ！

RACSubject *a=[RACSubject  subject];
RACCompoundDisposable *disposable =   [RACCompoundDisposable compoundDisposable] ;
RACPassthroughSubscriber *subscriber = [[RACPassthroughSubscriber alloc]  initWithSubscriber:a  signal:a disposable:disposable]  ;
RACCompoundDisposable *disposable2=[RACCompoundDisposable compoundDisposable];
RACPassthroughSubscriber *subscriber2 = [[RACPassthroughSubscriber alloc]  initWithSubscriber:subscriber  signal:a disposable:disposable2]  ;
[a   subscribeNext:^(id  _Nullable x) {
    NSLog(@"222 ->   %@", x);
}];
[subscriber2  sendNext:@"1111"];

二、任何订阅者本质上都是RACPassthroughSubscriber。 都持有一个disposable。

其目的是在

RACSubject (@property (nonatomic, strong, readonly) NSMutableArray *subscribers;)
RACPassthroughSubscriber(@property (nonatomic, strong, readonly) id<RACSubscriber> innerSubscriber;)
RACSubscriber////innerSubscriber
再调用sendXXX 先判断是否disposable。 在sendNext时，判断disposable是否已经销毁了，销毁了则不执行block； 在sendError和sendCompleted时执行相关disposable完成在创建订阅者时预置的销毁操作。

1、RACSubject
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

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

2、RACDynamicSignal

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;

三、RacDsiposable类

RACDisposable

RACCompoundDisposable

RACKVOTrampoline

RACScopedDisposable

RACSerialDisposable

RACSerialDisposable 在Rac源代码出现：
1、RACMulticastConnection

if (shouldConnect) {
    self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
}

2、RacSignal.m

Bind方法中
selfDisposable.disposable = bindingDisposable;

3、RACSignal+Operations.m

1、throttle
nextDisposable.disposable = [delayScheduler afterDelay:interval schedule:^{
flushNext(YES);
}];

2、catch

        catchDisposable.disposable = [signal subscribe:subscriber];

3、bufferWithTime
if (values.count == 0) {
timerDisposable.disposable = [scheduler afterDelay:interval schedule:flushValues];
}

4、flatten
serialDisposable.disposable = [signal subscribeNext:^(id x) {
[subscriber sendNext:x];
} error:^(NSError *error) {
[subscriber sendError:error];
} completed:^{

5、takeUntilReplacement
if (!selfDisposable.disposed) {
selfDisposable.disposable = [[self
concat:[RACSignal never]]
subscribe:subscriber];
}

6、sample

    samplerDisposable.disposable = [sampler subscribeNext:^(id _) {
        BOOL shouldSend = NO;
        id value;
        [lock lock];
        shouldSend = hasValue;
        value = lastValue;
        [lock unlock];

        if (shouldSend) {
            [subscriber sendNext:value];
        }
    } error:^(NSError *error) {
        [sourceDisposable dispose];
        [subscriber sendError:error];
    } completed:^{
        [sourceDisposable dispose];
        [subscriber sendCompleted];
    }];