日常使用
RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber){
/// STEP 1
[subscriber sendNext:@(1)];
[subscriber sendNext:@(2)];
[subscriber sendNext:@(3)];
[subscriber sendCompleted];
return [RACDisposable disposableWithBlock:^{
/// STEP 5
NSLog(@"signal dispose");
}];
}];
RACDisposable *disposable = [signal subscribeNext:^(id x) {
/// STEP 2
NSLog(@"received value = %@", x);
} error:^(NSError *error) {
/// STEP 3
NSLog(@"received error: %@", error);
} completed:^{
/// STEP 4
NSLog(@"received completed");
}];
[disposable dispose];
RACSignal 从创建到给订阅发送事件步骤归纳
- 调用RACSignal createSignal 的方法，返回子类 RACDynamicSignal 对象并保存闭包 didSubscribe
@implementation RACSignal
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
return [RACDynamicSignal createSignal:didSubscribe]; // 返回 RACDynamicSignal 类型的对象
}
@interface RACDynamicSignal : RACSignal
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe;
@end
@implementation RACDynamicSignal
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
RACDynamicSignal *signal = [[self alloc] init];
signal->_didSubscribe = [didSubscribe copy];
return [signal setNameWithFormat:@"+createSignal:"];
}
- 订阅信号，调用RACSignal subscribeNext等方法，在该方法中会创建 RACSubscriber 订阅者对象
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
NSCParameterAssert(nextBlock != NULL);
RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
return [self 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);
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;
}
- 步骤1创建的 RACDynamicSignal 调用 subscribe 方法，创建 RACCompoundDisposable 和 RACPassthroughSubscriber 对象，RACPassthroughSubscriber 会保存 步骤1返回的 RACDynamicSignal（signal属性），步骤2中的 RACSubscriber（innerSubscriber属性）
@implementation RACDynamicSignal
- (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:^{
//触发block，执行后续的信号发送操作，并注册丢弃信号的响应事件
RACDisposable *innerDisposable = self.didSubscribe(subscriber);
[disposable addDisposable:innerDisposable];
}];
[disposable addDisposable:schedulingDisposable];
}
return disposable;
}
- 执行 RACDynamicSignal 保存的 didSubscribe 闭包，闭包内调用 RACPassthroughSubscriber sendNext，sendError，sendCompleted 方法。
@implementation RACPassthroughSubscriber
- (void)sendNext:(id)value {
if (self.disposable.disposed) return;
if (RACSIGNAL_NEXT_ENABLED()) {
RACSIGNAL_NEXT(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description), cleanedDTraceString([value description]));
}
[self.innerSubscriber sendNext:value];
}
- (void)sendError:(NSError *)error {
if (self.disposable.disposed) return;
if (RACSIGNAL_ERROR_ENABLED()) {
RACSIGNAL_ERROR(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description), cleanedDTraceString(error.description));
}
[self.innerSubscriber sendError:error];
}
- (void)sendCompleted {
if (self.disposable.disposed) return;
if (RACSIGNAL_COMPLETED_ENABLED()) {
RACSIGNAL_COMPLETED(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description));
}
[self.innerSubscriber sendCompleted];
}
- RACPassthroughSubscriber 执行保存的 innerSubscriber 对应的 sendNext，sendError，sendCompleted 方法
@implementation 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();
}
}
