RACSubscriber.m
2.48 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
//
// RACSubscriber.m
// ReactiveCocoa
//
// Created by Josh Abernathy on 3/1/12.
// Copyright (c) 2012 GitHub, Inc. All rights reserved.
//
#import "RACSubscriber.h"
#import "RACSubscriber+Private.h"
#import "RACEXTScope.h"
#import "RACCompoundDisposable.h"
@interface RACSubscriber ()
// 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 RACSubscriber
#pragma mark Lifecycle
+ (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;
}
- (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