ReactiveCocoa 学习笔记
先贴上我看的博客，大部分内容来自以下博客
学的时候最好先建个工程，敲个Demo,还可以及时跳进去研究一下RAC的代码，思路会清晰很多
什么是Functional Reactive Programming
Functional Reactive Programming(以下简称FRP)是一种响应变化的编程范式。先来看一小段代码
a = 2
b = 2
c = a + b // c is 4
b = 3
now what is the value of c?
如果使用FRP，c 的值将会随着 b 的值改变而改变，所以叫做「响应式编程」。比较直观的例子就是Excel，当改变某一个单元格的内容时，该单元格相关的计算结果也会随之改变。
image
就以上两篇文章总结一下知识点,简单说下RAC的基本使用
1.创建信号：signal = [RACSignal createSignal:^RACDisposable *(id subscriber) {//订阅时执行
[subscriber sendError:accessError]; //出现错误
[subscriber sendNext:obj]; //通知订阅者来操作obj
[subscriber sendCompleted]; //完成发送，移除订阅者
}]
通过一些UI类别可以快速得到一个信号以供订阅
如 TextField.rac_textSignal
Button.rac_signalForControlEvents:UIControlEventTouchUpInside
在对应的空间后面输入rac_就会有提示。还可以这样RAC(self.titleLabel,text，@“没有的话就是这个默认值”) = TextField.rac_textSignal
2.订阅信号：subscibe: [siganal subscribeNext:^(id x){处理最近的next事件}error:{处理最先发生的error事件}]
3.筛选事件：
- filter: 把信号进行筛选，满足筛选条件的才会传给订阅者，如果用了combine,会保留最新满足的值。衍生函数有self.inputTextField.rac_textSignal ignore:@"sunny",还有 ignoreValues 这个比较极端，忽略所有值，只关心Signal结束，也就是只取Comletion和Error两个消息，中间所有值都丢弃.
- distinctUntilChanged ：不是filter 的衍生， 它将这一次的值与上一次做比较，当相同时（也包括- isEqual:）被忽略掉 [RACObserve(self.user, username) distinctUntilChanged]
还有 take（取） skip(跳)
- take: (NSUInteger)
从开始一共取N次的next值，不包括Competion和Error
- takeLast: (NSUInteger)
取最后N次的next值，注意，由于一开始不能知道这个Signal将有多少个next值，所以RAC实现它的方法是将所有next值都存起来，然后原Signal完成时再将后N个依次发送给接收者，但Error发生时依然是立刻发送的。
- takeUntil:(RACSignal *) 当给定的signal sendNext时，当前的signal就sendCompleted,把这个管道封住。例如cell的重用，cell上面有个btn,每次重用的时候，要把之前的btn相关的信号的水龙头去掉，再重新订阅。cell.detailButton rac_signalForControlEvents:UIControlEventTouchUpInside] takeUntil:cell.rac_prepareForReuseSignal
- takeUntilBlock:(BOOL (^)(id x))
对于每个next值，运行block，当block返回YES时停止取值
[self.inputTextField.rac_textSignal takeUntilBlock:^BOOL(NSString *value) {
return [value isEqualToString:@"stop"]
- takeWhileBlock:(BOOL (^)(id x))
上面的反向逻辑，对于每个next值，block返回 YES时才取值
- skip:(NSUInteger)
从开始跳过N次的next值
- skipUntilBlock:(BOOL (^)(id x))
和- takeUntilBlock:同理，一直跳，直到block为YES
- skipWhileBlock:(BOOL (^)(id x))
和- takeWhileBlock:同理，一直跳，直到block为NO
[[self.usernameTextField.rac_textSignal
filter:^BOOL(id value){
NSString*text = value;
return text.length > 3;
}]
subscribeNext:^(id x){
NSLog(@"%@", x);
}];
4.映射转换：map: 将信号传过来的值转换成自己想要的值，然后再传给订阅者。（函数式编程就是面对值的编程）
[[[self.usernameTextField.rac_textSignal
map:^id(NSString*text){
return @(text.length);
}]
filter:^BOOL(NSNumber*length){
return[length integerValue] > 3;
}]
subscribeNext:^(id x){
NSLog(@"%@", x);
}];
5.多信号绑定 combine:
RACSignal *signUpActiveSignal =
[RACSignal combineLatest:@[validUsernameSignal, validPasswordSignal]
reduce:^id(NSNumber*usernameValid, NSNumber *passwordValid){
return @([usernameValid boolValue]&&[passwordValid boolValue]);
}];
combineLatest:reduce:方法把signal1和signal2产生的最新的值聚合在一起，当两个信号都有值的时候会合并生成一个新的信号。每次这两个源信号的任何一个产生新值时，reduce block都会执行,**如果中间其中一个有fillter 且没有满足条件，那么会保留最近满足条件的那个值**，block的返回值会发给下一个信号。
** 这里解释一下signal 的分割和聚合 **
** 1. 分割——信号可以有很多subscriber，也就是作为很多后续步骤的源。注意上图中那个用来表示用户名和密码有效性的布尔信号，它被分割成多个，用于不同的地方。**
** 2. 聚合——多个信号可以聚合成一个新的信号，在上面的例子中，两个布尔信号聚合成了一个。实际上你可以聚合并产生任何类型的信号。 **
6.取双层信号中内层信号的值：flattenMap: 将异步请求包装在signal里面 再通过flattenMap 拿到信号中的信号的值作为往下传递的值。
flattenMap方法通过调用block（value）来创建一个新的方法，它可以灵活的定义新创建的信号。
map方法，将会创建一个和原来一模一样的信号，只不过新的信号传递的值变为了block（value）。
** map创建一个新的信号，信号的value是block(value)，也就是说，如果block(value)是一个信号，那么就是信号的value仍然是信号。如果是flattenMap则会继续调用这个信号的value，作为新的信号的value。 **
[[[self.signInButton
rac_signalForControlEvents:UIControlEventTouchUpInside]
flattenMap:^id(id x){
return[self signInSignal];
}]
subscribeNext:^(id x){
NSLog(@"Sign in result: %@", x);
}];
7.添加附加操作 doNext: （side - effects）函数式编程是不建议在函数操作的过程中，改变事件本身。如我们在点击按钮后，让其在请求完成之前置为不可用状态。
[[[[self.signInButton
rac_signalForControlEvents:UIControlEventTouchUpInside]
doNext:^(id x){
self.signInButton.enabled =NO;
self.signInFailureText.hidden =YES;
}]
flattenMap:^id(id x){
return[self signInSignal];
}]
subscribeNext:^(NSNumber*signedIn){
self.signInButton.enabled =YES;
BOOL success =[signedIn boolValue];
self.signInFailureText.hidden = success;
if(success){
[self performSegueWithIdentifier:@"signInSuccess" sender:self];
}
}];
8.RAC宏允许直接把信号的输出的值应用到对象的属性上。RAC宏有两个参数，第一个是需要设置属性值的对象，第二个是属性名。每次信号产生一个next事件，传递过来的值都会应用到该属性上。
9.then 会等待上一个signal中completed事件的发送，然后再订阅then block 返回的block.这样就高效的把控制权从一个signal 传给了下一个。当然error事件会跳过then方法，直接执行subscribeNext:error的中error事件
10.deliverOn RACScheduler 线程 对GCD的简单封装
11.throttle：1 节流，拿到传过来的signal后产生一个新的信号，在间隔1s的时间内如果有新的signal流进来，那么就保留最新的，如果1s内没有新的signal了，就发送next事件，继续往下走。适合在类似地点搜索，快速的网络搜索
上面任何的信号转换都是拿到原有信号再产生新的信号,注意下新信号对原有信号订阅的时机是在新信号被订阅的时候才会去订阅源信号。新信号生成同样调用createSignal方法，在didsubscribe回调中对原有信号进行订阅,当最后生成的新的信号被订阅的时候（subscribe:）会调用自己的didsubscribe,然后订阅原有信号，执行原有信号的didsubscribe:
可以这样理解:一个水管接另一个水管，但都是结冰的死的冷信号，只有最后一个被subscribe了，即装上了水龙头，才算有了出口，这样就打通了整个管道，变成流动的活的热信号。
12.关于冷信号的副作用，以及冷信号与热信号之间的转换signal - subject
热信号是主动的，即使你没有订阅事件，它仍然会时刻推送。而冷信号是被动的，只有当你订阅的时候，它才会发送消息。
热信号可以有多个订阅者，是一对多，信号可以与订阅者共享信息。而冷信号只能一对一，当有不同的订阅者，消息会从新完整发送。
先看下冷信号的一般实现步骤
步骤一：[RACSignal createSignal]来获得signal
RACSignal.m中：
+ ( RACSignal *)createSignal:( RACDisposable * (^)( id < RACSubscriber > subscriber))didSubscribe {
return [ RACDynamicSignal createSignal :didSubscribe];
}
RACDynamicSignal.m中
+ ( RACSignal *)createSignal:( RACDisposable * (^)( id < RACSubscriber > subscriber))didSubscribe {
RACDynamicSignal *signal = [[ self alloc ] init ];
signal-> _didSubscribe = [didSubscribe copy ];
return [signal setNameWithFormat : @"+createSignal:" ];
}
[RACSignal createSignal]会调用子类RACDynamicSignal的createSignal来返回一个signal，并在signal中保存后面的 didSubscribe这个block
步骤二：订阅信号 [signal subscribeNext:]来获得subscriber，然后进行subscription
RACSignal.m中：
- ( RACDisposable *)subscribeNext:( void (^)( id x))nextBlock {
RACSubscriber *o = [ RACSubscriber subscriberWithNext :nextBlock error : NULL completed : NULL ];
return [ self subscribe :o];//重点，信号被订阅
}
RACSubscriber.m中：
+ ( 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;
}
RACDynamicSignal.m中：
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
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;
}
[signal subscribeNext]先会获得一个subscriber，这个subscriber中保存了nextBlock、errorBlock、completedBlock
由于这个signal其实是RACDynamicSignal类型的，这个[self subscribe]方法会调用步骤一中保存的didSubscribe，参数就是1中的subscriber
步骤三：进入didSubscribe，通过[subscriber sendNext:]来执行next block
RACSubscriber.m中：
- (void)sendNext:(id)value {
@synchronized (self) {
void (^nextBlock)(id) = [self.next copy];
if (nextBlock == nil) return;
nextBlock(value);
}
}
举个栗子：
self.sessionManager = [[AFHTTPSessionManager alloc] initWithBaseURL:[NSURL URLWithString:@"http://api.xxxx.com"]];
self.sessionManager.requestSerializer = [AFJSONRequestSerializer serializer];
self.sessionManager.responseSerializer = [AFJSONResponseSerializer serializer];
@weakify(self)
RACSignal *fetchData = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
@strongify(self)
NSURLSessionDataTask *task = [self.sessionManager GET:@"fetchData" parameters:@{@"someParameter": @"someValue"} success:^(NSURLSessionDataTask *task, id responseObject) {
[subscriber sendNext:responseObject];
[subscriber sendCompleted];
} failure:^(NSURLSessionDataTask *task, NSError *error) {
[subscriber sendError:error];
}];
return [RACDisposable disposableWithBlock:^{
if (task.state != NSURLSessionTaskStateCompleted) {
[task cancel];
}
}];
}];
RACSignal *title = [fetchData flattenMap:^RACSignal *(NSDictionary *value) {
if ([value[@"title"] isKindOfClass:[NSString class]]) {
return [RACSignal return:value[@"title"]];
} else {
return [RACSignal error:[NSError errorWithDomain:@"some error" code:400 userInfo:@{@"originData": value}]];
}
}];
RACSignal *desc = [fetchData flattenMap:^RACSignal *(NSDictionary *value) {
if ([value[@"desc"] isKindOfClass:[NSString class]]) {
return [RACSignal return:value[@"desc"]];
} else {
return [RACSignal error:[NSError errorWithDomain:@"some error" code:400 userInfo:@{@"originData": value}]];
}
}];
RACSignal *renderedDesc = [desc flattenMap:^RACStream *(NSString *value) {
NSError *error = nil;
RenderManager *renderManager = [[RenderManager alloc] init];
NSAttributedString *rendered = [renderManager renderText:value error:&error];
if (error) {
return [RACSignal error:error];
} else {
return [RACSignal return:rendered];
}
}];
RAC(self.someLablel, text) = [[title catchTo:[RACSignal return:@"Error"]] startWith:@"Loading..."];
RAC(self.originTextView, text) = [[desc catchTo:[RACSignal return:@"Error"]] startWith:@"Loading..."];
RAC(self.renderedTextView, attributedText) = [[renderedDesc catchTo:[RACSignal return:[[NSAttributedString alloc] initWithString:@"Error"]]] startWith:[[NSAttributedString alloc] initWithString:@"Loading..."]];
[[RACSignal merge:@[title, desc, renderedDesc]] subscribeError:^(NSError *error) {
UIAlertView *alertView = [[UIAlertView alloc] initWithTitle:@"Error" message:error.domain delegate:nil cancelButtonTitle:@"OK" otherButtonTitles:nil];
[alertView show];
}];
从上面可以看到 fatchData 被间接订阅了6次，那么它的didSubscribe也会走6次，也就是请求6次。这就是冷信号的副作用，每次订阅都重新计算，而在函数编程中 纯函数的调用在相同参数下的返回值第二次不需要计算。在oc中怎么做到，是不是可以做一些缓冲，或保存一下计算结果，自己控制需不需要重新计算。
RAC规避副作用的做法就是将冷信号转成热信号。冷信号与热信号的本质区别在于是否保持状态，冷信号的多次订阅是不保持状态的，而热信号的多次订阅可以保持状态。所以一种将冷信号转换为热信号的方法就是，将冷信号订阅，订阅到的每一个时间通过RACSbuject发送出去，其他订阅者只订阅这个RACSubject。
以下这些就是冷信号到热信号的转变
RACSignal+Operation.h中
这5个方法中，最为重要的就是- (RACMulticastConnection *)multicast:(RACSubject *)subject;这个方法了，其他几个方法也是间接调用它的。
- (RACMulticastConnection *)publish;
- (RACMulticastConnection *)multicast:(RACSubject *)subject;
- (RACSignal *)replay;
- (RACSignal *)replayLast;
- (RACSignal *)replayLazily;
RACMulticastConnection.m中：
/// implementation RACSignal (Operations)
- (RACMulticastConnection *)multicast:(RACSubject *)subject {
[subject setNameWithFormat:@"[%@] -multicast: %@", self.name, subject.name];
RACMulticastConnection *connection = [[RACMulticastConnection alloc] initWithSourceSignal:self subject:subject];
return connection;
}
- (id)initWithSourceSignal:(RACSignal *)source subject:(RACSubject *)subject {
NSCParameterAssert(source != nil);
NSCParameterAssert(subject != nil);
self = [super init];
if (self == nil) return nil;
_sourceSignal = source;
_serialDisposable = [[RACSerialDisposable alloc] init];
_signal = subject;
return self;
}
- (RACDisposable *)connect {
BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);
if (shouldConnect) {
self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal]; //调用RACSignal的subcribe:真正的订阅。
}
return self.serialDisposable;
}
- (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];
}
- 当RACSignal类的实例调用- (RACMulticastConnection *)multicast:(RACSubject *)subject时，以self和subject作为构造参数创建一个RACMulticastConnection实例。
- RACMulticastConnection构造的时候，保存source和subject作为成员变量，创建一个RACSerialDisposable对象，用于取消订阅。
- 当RACMulticastConnection类的实例调用- (RACDisposable *)connect这个方法的时候，判断是否是第一次。如果是的话用_signal这个成员变量来订阅sourceSignal之后返回self.serialDisposable；否则直- 接返回self.serialDisposable。这里面订阅sourceSignal是重点。
- RACMulticastConnection的signal只读属性，就是一个热信号，订阅这个热信号就避免了各种副作用的问题。它会在- (RACDisposable *)connect第一次调用后，根据sourceSignal的订阅结果来传递事件。
- 想要确保第一次订阅就能成功订阅sourceSignal，可以使用- (RACSignal *)autoconnect这个方法，它保证了第一个订阅者触发sourceSignal的订阅，也保证了当返回的信号所有订阅者都关闭连接后sourceSignal被正确关闭连接。
下面再来看看其他几个方法的实现：
/// implementation RACSignal (Operations)
- (RACMulticastConnection *)publish {
RACSubject *subject = [[RACSubject subject] setNameWithFormat:@"[%@] -publish", self.name];
RACMulticastConnection *connection = [self multicast:subject];
return connection;
}
- (RACSignal *)replay {
RACReplaySubject *subject = [[RACReplaySubject subject] setNameWithFormat:@"[%@] -replay", self.name];
RACMulticastConnection *connection = [self multicast:subject];
[connection connect];
return connection.signal;
}
- (RACSignal *)replayLast {
RACReplaySubject *subject = [[RACReplaySubject replaySubjectWithCapacity:1] setNameWithFormat:@"[%@] -replayLast", self.name];
RACMulticastConnection *connection = [self multicast:subject];
[connection connect];
return connection.signal;
}
- (RACSignal *)replayLazily {
RACMulticastConnection *connection = [self multicast:[RACReplaySubject subject]];
return [[RACSignal
defer:^{
[connection connect];
return connection.signal;
}]
setNameWithFormat:@"[%@] -replayLazily", self.name];
}
这几个方法的实现都相当简单，只是为了简化而封装，具体说明一下：
-
(RACMulticastConnection *)publish就是帮忙创建了RACSubject。
-
(RACSignal *)replay就是用RACReplaySubject来作为subject，并立即执行connect操作，返回connection.signal。其作用是上面提到的replay功能，即后来的订阅者可以收到历史值。
-
(RACSignal *)replayLast就是用Capacity为1的RACReplaySubject来替换- (RACSignal *)replay的`subject。其作用是使后来订阅者只收到最后的历史值。
-
(RACSignal *)replayLazily和- (RACSignal *)replay的区别就是replayLazily会在第一次订阅的时候才订阅sourceSignal。
-
到这里，就清楚了，咱们只需要把之前请求的fatchData = [fatchData replayLazily] 就好了。
看下subject 的实现细节
- (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];
}
return [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];
}
}];
}
从subscribe:的实现可以看出，对RACSubject对象的每次subscription，都是将这个subscriber加到subscribers数组中，并没有调用 didSubScirbe（）
对比下RACDynamicSignal的subscibe:
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
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;
}
其中会调用 self.didSubscribe(subscriber)，调用了 这个也就会调用其中的sendnext.
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
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;
}
subject的 sendNext:
- (void)sendNext:(id)value {
[self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
[subscriber sendNext:value];
}];
}
RACReplaySubject
RACReplaySubject.m中：
- (instancetype)initWithCapacity:(NSUInteger)capacity {
self = [super init];
if (self == nil) return nil;
_capacity = capacity;
_valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
return self;
}
从init中我们看出，RACReplaySubject对象持有capacity变量（用于决定valuesReceived缓存多少个sendNext：出来的value，这在区分replay和replayLast的时候特别有用）以及valuesReceived数组（用来保存sendNext:出来的value），这二者接下来会重点涉及到
- (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;
从subscribe:可以看出，RACReplaySubject对象每次subscription，都会把之前valuesReceived中buffer的value重新sendNext一遍，然后调用super把当前的subscriber加入到subscribers数组中
- (void)sendNext:(id)value {
@synchronized (self) {
[self.valuesReceived addObject:value ?: RACTupleNil.tupleNil];
[super sendNext:value];
if (self.capacity != RACReplaySubjectUnlimitedCapacity && self.valuesReceived.count > self.capacity) {
[self.valuesReceived removeObjectsInRange:NSMakeRange(0, self.valuesReceived.count - self.capacity)];
}
}
}
从sendNext:可以看出，RACReplaySubject对象会buffer每次sendNext的value，然后会调用super，对subscribers中的每个subscriber，调用sendNext。buffer的数量是根据self.capacity来决定的。
