31
31
* the type of items the Subscriber expects to observe
32
32
*/
33
33
public abstract class Subscriber <T > implements Observer <T >, Subscription {
34
+
35
+ // represents requested not set yet
36
+ private static final Long NOT_SET = Long .MIN_VALUE ;
34
37
35
- private final SubscriptionList cs ;
36
- private final Subscriber <?> op ;
38
+ private final SubscriptionList subscriptions ;
39
+ private final Subscriber <?> subscriber ;
37
40
/* protected by `this` */
38
- private Producer p ;
41
+ private Producer producer ;
39
42
/* protected by `this` */
40
- private long requested = Long . MIN_VALUE ; // default to not set
43
+ private long requested = NOT_SET ; // default to not set
41
44
42
45
protected Subscriber () {
43
46
this (null , false );
44
47
}
45
48
46
- protected Subscriber (Subscriber <?> op ) {
47
- this (op , true );
49
+ protected Subscriber (Subscriber <?> subscriber ) {
50
+ this (subscriber , true );
48
51
}
49
52
50
53
/**
@@ -53,15 +56,15 @@ protected Subscriber(Subscriber<?> op) {
53
56
* <p>
54
57
* To retain the chaining of subscribers, add the created instance to {@code op} via {@link #add}.
55
58
*
56
- * @param op
59
+ * @param subscriber
57
60
* the other Subscriber
58
61
* @param shareSubscriptions
59
62
* {@code true} to share the subscription list in {@code op} with this instance
60
63
* @since 1.0.6
61
64
*/
62
- protected Subscriber (Subscriber <?> op , boolean shareSubscriptions ) {
63
- this .op = op ;
64
- this .cs = shareSubscriptions && op != null ? op . cs : new SubscriptionList ();
65
+ protected Subscriber (Subscriber <?> subscriber , boolean shareSubscriptions ) {
66
+ this .subscriber = subscriber ;
67
+ this .subscriptions = shareSubscriptions && subscriber != null ? subscriber . subscriptions : new SubscriptionList ();
65
68
}
66
69
67
70
/**
@@ -73,12 +76,12 @@ protected Subscriber(Subscriber<?> op, boolean shareSubscriptions) {
73
76
* the {@code Subscription} to add
74
77
*/
75
78
public final void add (Subscription s ) {
76
- cs .add (s );
79
+ subscriptions .add (s );
77
80
}
78
81
79
82
@ Override
80
83
public final void unsubscribe () {
81
- cs .unsubscribe ();
84
+ subscriptions .unsubscribe ();
82
85
}
83
86
84
87
/**
@@ -88,7 +91,7 @@ public final void unsubscribe() {
88
91
*/
89
92
@ Override
90
93
public final boolean isUnsubscribed () {
91
- return cs .isUnsubscribed ();
94
+ return subscriptions .isUnsubscribed ();
92
95
}
93
96
94
97
/**
@@ -124,57 +127,64 @@ protected final void request(long n) {
124
127
if (n < 0 ) {
125
128
throw new IllegalArgumentException ("number requested cannot be negative: " + n );
126
129
}
127
- Producer shouldRequest = null ;
130
+
131
+ // if producer is set then we will request from it
132
+ // otherwise we increase the requested count by n
133
+ Producer producerToRequestFrom = null ;
128
134
synchronized (this ) {
129
- if (p != null ) {
130
- shouldRequest = p ;
131
- } else if (requested == Long .MIN_VALUE ) {
132
- requested = n ;
133
- } else {
134
- final long total = requested + n ;
135
- // check if overflow occurred
136
- if (total < 0 ) {
137
- requested = Long .MAX_VALUE ;
138
- } else {
139
- requested = total ;
140
- }
135
+ if (producer != null ) {
136
+ producerToRequestFrom = producer ;
137
+ } else {
138
+ addToRequested (n );
139
+ return ;
141
140
}
142
141
}
143
- // after releasing lock
144
- if (shouldRequest != null ) {
145
- shouldRequest .request (n );
146
- }
142
+ // after releasing lock (we should not make requests holding a lock)
143
+ producerToRequestFrom .request (n );
147
144
}
148
145
146
+ private void addToRequested (long n ) {
147
+ if (requested == NOT_SET ) {
148
+ requested = n ;
149
+ } else {
150
+ final long total = requested + n ;
151
+ // check if overflow occurred
152
+ if (total < 0 ) {
153
+ requested = Long .MAX_VALUE ;
154
+ } else {
155
+ requested = total ;
156
+ }
157
+ }
158
+ }
159
+
149
160
/**
150
161
* @warn javadoc description missing
151
162
* @warn param producer not described
152
- * @param producer
163
+ * @param p
153
164
*/
154
- public void setProducer (Producer producer ) {
165
+ public void setProducer (Producer p ) {
155
166
long toRequest ;
156
- boolean setProducer = false ;
167
+ boolean passToSubscriber = false ;
157
168
synchronized (this ) {
158
169
toRequest = requested ;
159
- p = producer ;
160
- if (op != null ) {
170
+ producer = p ;
171
+ if (subscriber != null ) {
161
172
// middle operator ... we pass thru unless a request has been made
162
- if (toRequest == Long . MIN_VALUE ) {
173
+ if (toRequest == NOT_SET ) {
163
174
// we pass-thru to the next producer as nothing has been requested
164
- setProducer = true ;
175
+ passToSubscriber = true ;
165
176
}
166
-
167
177
}
168
178
}
169
179
// do after releasing lock
170
- if (setProducer ) {
171
- op .setProducer (p );
180
+ if (passToSubscriber ) {
181
+ subscriber .setProducer (producer );
172
182
} else {
173
183
// we execute the request with whatever has been requested (or Long.MAX_VALUE)
174
- if (toRequest == Long . MIN_VALUE ) {
175
- p .request (Long .MAX_VALUE );
184
+ if (toRequest == NOT_SET ) {
185
+ producer .request (Long .MAX_VALUE );
176
186
} else {
177
- p .request (toRequest );
187
+ producer .request (toRequest );
178
188
}
179
189
}
180
190
}
0 commit comments