You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null`
20
+
if (s == null) thrownull;
21
+
22
+
if (subscription != null) { // If someone has made a mistake and added this Subscriber multiple times, let's handle it gracefully
23
+
try {
24
+
s.cancel(); // Cancel the additional subscription
25
+
} catch(finalThrowablet) {
26
+
//Subscription.cancel is not allowed to throw an exception, according to rule 3.15
27
+
(newIllegalStateException(s + " violated the Reactive Streams rule 3.15 by throwing an exception from cancel.", t)).printStackTrace(System.err);
28
+
}
29
+
} else {
30
+
// We have to assign it locally before we use it, if we want to be a synchronous `Subscriber`
31
+
// Because according to rule 3.10, the Subscription is allowed to call `onNext` synchronously from within `request`
32
+
subscription = s;
33
+
}
34
+
}
35
+
36
+
/**
37
+
* Requests the provided number of elements from the `Subscription` of this `Subscriber`.
38
+
* NOTE: This makes no attempt at thread safety so only invoke it once from the outside to initiate the demand.
39
+
* @return `true` if successful and `false` if not (either due to no `Subscription` or due to exceptions thrown)
40
+
*/
41
+
publicbooleantriggerDemand(finallongn) {
42
+
finalSubscriptions = subscription;
43
+
if (s == null) returnfalse;
44
+
else {
45
+
try {
46
+
s.request(n);
47
+
} catch(finalThrowablet) {
48
+
// Subscription.request is not allowed to throw according to rule 3.16
49
+
(newIllegalStateException(s + " violated the Reactive Streams rule 3.16 by throwing an exception from request.", t)).printStackTrace(System.err);
50
+
returnfalse;
51
+
}
52
+
returntrue;
53
+
}
54
+
}
55
+
56
+
@OverridepublicvoidonNext(finalTelement) {
57
+
if (subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec
58
+
(newIllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onNext prior to onSubscribe.")).printStackTrace(System.err);
59
+
} else {
60
+
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null`
61
+
if (element == null) thrownull;
62
+
63
+
if (!done) { // If we aren't already done
64
+
try {
65
+
finallongneed = foreach(element);
66
+
if (need > 0) triggerDemand(need);
67
+
elseif (need == 0) {}
68
+
else {
69
+
done();
70
+
}
71
+
} catch (finalThrowablet) {
72
+
done();
73
+
try {
74
+
onError(t);
75
+
} catch (finalThrowablet2) {
76
+
//Subscriber.onError is not allowed to throw an exception, according to rule 2.13
77
+
(newIllegalStateException(this + " violated the Reactive Streams rule 2.13 by throwing an exception from onError.", t2)).printStackTrace(System.err);
78
+
}
79
+
}
80
+
}
81
+
}
82
+
}
83
+
84
+
// Showcases a convenience method to idempotently marking the Subscriber as "done", so we don't want to process more elements
85
+
// herefor we also need to cancel our `Subscription`.
86
+
privatevoiddone() {
87
+
//On this line we could add a guard against `!done`, but since rule 3.7 says that `Subscription.cancel()` is idempotent, we don't need to.
88
+
done = true; // If we `foreach` throws an exception, let's consider ourselves done (not accepting more elements)
89
+
try {
90
+
subscription.cancel(); // Cancel the subscription
91
+
} catch(finalThrowablet) {
92
+
//Subscription.cancel is not allowed to throw an exception, according to rule 3.15
93
+
(newIllegalStateException(subscription + " violated the Reactive Streams rule 3.15 by throwing an exception from cancel.", t)).printStackTrace(System.err);
94
+
}
95
+
}
96
+
97
+
// This method is left as an exercise to the reader/extension point
98
+
// Don't forget to call `triggerDemand` at the end if you are interested in more data,
99
+
// a return value of < 0 indicates that the subscription should be cancelled,
100
+
// a value of 0 indicates that there is no current need,
101
+
// a value of > 0 indicates the current need.
102
+
protectedabstractlongforeach(finalTelement);
103
+
104
+
@OverridepublicvoidonError(finalThrowablet) {
105
+
if (subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec
106
+
(newIllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onError prior to onSubscribe.")).printStackTrace(System.err);
107
+
} else {
108
+
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null`
109
+
if (t == null) thrownull;
110
+
// Here we are not allowed to call any methods on the `Subscription` or the `Publisher`, as per rule 2.3
111
+
// And anyway, the `Subscription` is considered to be cancelled if this method gets called, as per rule 2.4
112
+
}
113
+
}
114
+
115
+
@OverridepublicvoidonComplete() {
116
+
if (subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec
117
+
(newIllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onComplete prior to onSubscribe.")).printStackTrace(System.err);
118
+
} else {
119
+
// Here we are not allowed to call any methods on the `Subscription` or the `Publisher`, as per rule 2.3
120
+
// And anyway, the `Subscription` is considered to be cancelled if this method gets called, as per rule 2.4
0 commit comments