Skip to content

Commit 7e89c1f

Browse files
vanniktechakarnokd
vanniktech
authored and
akarnokd
committed
2.x: Add Observable.rangeLong & Flowable.rangeLong (#4687)
* 2.x: Add Observable.rangeLong & Flowable.rangeLong * Clean up cast * Adjust Long overflow checks * Add test for rangeLong count 1 * Fix ObservableRangeLongTest.testRangeWithOverflow5
1 parent 5888b23 commit 7e89c1f

File tree

6 files changed

+918
-0
lines changed

6 files changed

+918
-0
lines changed

‎src/main/java/io/reactivex/Flowable.java

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3455,6 +3455,50 @@ public static Flowable<Integer> range(int start, int count) {
34553455
return RxJavaPlugins.onAssembly(new FlowableRange(start, count));
34563456
}
34573457

3458+
/**
3459+
* Returns a Flowable that emits a sequence of Longs within a specified range.
3460+
* <p>
3461+
* <img width="640" height="195" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/range.png" alt="">
3462+
* <dl>
3463+
* <dt><b>Backpressure:</b></dt>
3464+
* <dd>The operator honors backpressure from downstream and signals values on-demand (i.e., when requested).</dd>
3465+
* <dt><b>Scheduler:</b></dt>
3466+
* <dd>{@code rangeLong} does not operate by default on a particular {@link Scheduler}.</dd>
3467+
* </dl>
3468+
*
3469+
* @param start
3470+
* the value of the first Long in the sequence
3471+
* @param count
3472+
* the number of sequential Longs to generate
3473+
* @return a Flowable that emits a range of sequential Longs
3474+
* @throws IllegalArgumentException
3475+
* if {@code count} is less than zero, or if {@code start} + {@code count} &minus; 1 exceeds
3476+
* {@code Long.MAX_VALUE}
3477+
* @see <a href="http://reactivex.io/documentation/operators/range.html">ReactiveX operators documentation: Range</a>
3478+
*/
3479+
@BackpressureSupport(BackpressureKind.FULL)
3480+
@SchedulerSupport(SchedulerSupport.NONE)
3481+
public static Flowable<Long> rangeLong(long start, long count) {
3482+
if (count < 0) {
3483+
throw new IllegalArgumentException("count >= 0 required but it was " + count);
3484+
}
3485+
3486+
if (count == 0) {
3487+
return empty();
3488+
}
3489+
3490+
if (count == 1) {
3491+
return just(start);
3492+
}
3493+
3494+
long end = start + (count - 1);
3495+
if (start > 0 && end < 0) {
3496+
throw new IllegalArgumentException("Overflow! start + count is bigger than Long.MAX_VALUE");
3497+
}
3498+
3499+
return RxJavaPlugins.onAssembly(new FlowableRangeLong(start, count));
3500+
}
3501+
34583502
/**
34593503
* Returns a Flowable that emits a Boolean value that indicates whether two Publisher sequences are the
34603504
* same by comparing the items emitted by each Publisher pairwise.

‎src/main/java/io/reactivex/Observable.java

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2992,6 +2992,47 @@ public static Observable<Integer> range(final int start, final int count) {
29922992
return RxJavaPlugins.onAssembly(new ObservableRange(start, count));
29932993
}
29942994

2995+
/**
2996+
* Returns an Observable that emits a sequence of Longs within a specified range.
2997+
* <p>
2998+
* <img width="640" height="195" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/range.png" alt="">
2999+
* <dl>
3000+
* <dt><b>Scheduler:</b></dt>
3001+
* <dd>{@code rangeLong} does not operate by default on a particular {@link Scheduler}.</dd>
3002+
* </dl>
3003+
*
3004+
* @param start
3005+
* the value of the first Long in the sequence
3006+
* @param count
3007+
* the number of sequential Longs to generate
3008+
* @return an Observable that emits a range of sequential Longs
3009+
* @throws IllegalArgumentException
3010+
* if {@code count} is less than zero, or if {@code start} + {@code count} &minus; 1 exceeds
3011+
* {@code Long.MAX_VALUE}
3012+
* @see <a href="http://reactivex.io/documentation/operators/range.html">ReactiveX operators documentation: Range</a>
3013+
*/
3014+
@SchedulerSupport(SchedulerSupport.NONE)
3015+
public static Observable<Long> rangeLong(long start, long count) {
3016+
if (count < 0) {
3017+
throw new IllegalArgumentException("count >= 0 required but it was " + count);
3018+
}
3019+
3020+
if (count == 0) {
3021+
return empty();
3022+
}
3023+
3024+
if (count == 1) {
3025+
return just(start);
3026+
}
3027+
3028+
long end = start + (count - 1);
3029+
if (start > 0 && end < 0) {
3030+
throw new IllegalArgumentException("Overflow! start + count is bigger than Long.MAX_VALUE");
3031+
}
3032+
3033+
return RxJavaPlugins.onAssembly(new ObservableRangeLong(start, count));
3034+
}
3035+
29953036
/**
29963037
* Returns an Observable that emits a Boolean value that indicates whether two ObservableSource sequences are the
29973038
* same by comparing the items emitted by each ObservableSource pairwise.

‎src/main/java/io/reactivex/internal/operators/flowable/FlowableRangeLong.java

Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
/**
2+
* Copyright 2016 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.internal.operators.flowable;
15+
16+
import io.reactivex.Flowable;
17+
import io.reactivex.internal.fuseable.ConditionalSubscriber;
18+
import io.reactivex.internal.subscriptions.BasicQueueSubscription;
19+
import io.reactivex.internal.subscriptions.SubscriptionHelper;
20+
import io.reactivex.internal.util.BackpressureHelper;
21+
import org.reactivestreams.Subscriber;
22+
23+
/**
24+
* Emits a range of long values.
25+
*/
26+
public final class FlowableRangeLong extends Flowable<Long> {
27+
final long start;
28+
final long end;
29+
30+
public FlowableRangeLong(long start, long count) {
31+
this.start = start;
32+
this.end = start + count;
33+
}
34+
35+
@Override
36+
public void subscribeActual(Subscriber<? super Long> s) {
37+
if (s instanceof ConditionalSubscriber) {
38+
s.onSubscribe(new RangeConditionalSubscription(
39+
(ConditionalSubscriber<? super Long>)s, start, end));
40+
} else {
41+
s.onSubscribe(new RangeSubscription(s, start, end));
42+
}
43+
}
44+
45+
abstract static class BaseRangeSubscription extends BasicQueueSubscription<Long> {
46+
47+
private static final long serialVersionUID = -2252972430506210021L;
48+
49+
final long end;
50+
51+
long index;
52+
53+
volatile boolean cancelled;
54+
55+
BaseRangeSubscription(long index, long end) {
56+
this.index = index;
57+
this.end = end;
58+
}
59+
60+
@Override
61+
public final int requestFusion(int mode) {
62+
return mode & SYNC;
63+
}
64+
65+
@Override
66+
public final Long poll() {
67+
long i = index;
68+
if (i == end) {
69+
return null;
70+
}
71+
index = i + 1;
72+
return i;
73+
}
74+
75+
@Override
76+
public final boolean isEmpty() {
77+
return index == end;
78+
}
79+
80+
@Override
81+
public final void clear() {
82+
index = end;
83+
}
84+
85+
@Override
86+
public final void request(long n) {
87+
if (SubscriptionHelper.validate(n)) {
88+
if (BackpressureHelper.add(this, n) == 0L) {
89+
if (n == Long.MAX_VALUE) {
90+
fastPath();
91+
} else {
92+
slowPath(n);
93+
}
94+
}
95+
}
96+
}
97+
98+
@Override
99+
public final void cancel() {
100+
cancelled = true;
101+
}
102+
103+
104+
abstract void fastPath();
105+
106+
abstract void slowPath(long r);
107+
}
108+
109+
static final class RangeSubscription extends BaseRangeSubscription {
110+
111+
private static final long serialVersionUID = 2587302975077663557L;
112+
113+
final Subscriber<? super Long> actual;
114+
115+
RangeSubscription(Subscriber<? super Long> actual, long index, long end) {
116+
super(index, end);
117+
this.actual = actual;
118+
}
119+
120+
@Override
121+
void fastPath() {
122+
long f = end;
123+
Subscriber<? super Long> a = actual;
124+
125+
for (long i = index; i != f; i++) {
126+
if (cancelled) {
127+
return;
128+
}
129+
a.onNext(i);
130+
}
131+
if (cancelled) {
132+
return;
133+
}
134+
a.onComplete();
135+
}
136+
137+
@Override
138+
void slowPath(long r) {
139+
long e = 0;
140+
long f = end;
141+
long i = index;
142+
Subscriber<? super Long> a = actual;
143+
144+
for (;;) {
145+
146+
while (e != r && i != f) {
147+
if (cancelled) {
148+
return;
149+
}
150+
151+
a.onNext(i);
152+
153+
e++;
154+
i++;
155+
}
156+
157+
if (i == f) {
158+
if (!cancelled) {
159+
a.onComplete();
160+
}
161+
return;
162+
}
163+
164+
r = get();
165+
if (e == r) {
166+
index = i;
167+
r = addAndGet(-e);
168+
if (r == 0L) {
169+
return;
170+
}
171+
e = 0L;
172+
}
173+
}
174+
}
175+
}
176+
177+
static final class RangeConditionalSubscription extends BaseRangeSubscription {
178+
179+
180+
private static final long serialVersionUID = 2587302975077663557L;
181+
182+
final ConditionalSubscriber<? super Long> actual;
183+
184+
RangeConditionalSubscription(ConditionalSubscriber<? super Long> actual, long index, long end) {
185+
super(index, end);
186+
this.actual = actual;
187+
}
188+
189+
@Override
190+
void fastPath() {
191+
long f = end;
192+
ConditionalSubscriber<? super Long> a = actual;
193+
194+
for (long i = index; i != f; i++) {
195+
if (cancelled) {
196+
return;
197+
}
198+
a.tryOnNext(i);
199+
}
200+
if (cancelled) {
201+
return;
202+
}
203+
a.onComplete();
204+
}
205+
206+
@Override
207+
void slowPath(long r) {
208+
long e = 0;
209+
long f = end;
210+
long i = index;
211+
ConditionalSubscriber<? super Long> a = actual;
212+
213+
for (;;) {
214+
215+
while (e != r && i != f) {
216+
if (cancelled) {
217+
return;
218+
}
219+
220+
if (a.tryOnNext(i)) {
221+
e++;
222+
}
223+
224+
i++;
225+
}
226+
227+
if (i == f) {
228+
if (!cancelled) {
229+
a.onComplete();
230+
}
231+
return;
232+
}
233+
234+
r = get();
235+
if (e == r) {
236+
index = i;
237+
r = addAndGet(-e);
238+
if (r == 0) {
239+
return;
240+
}
241+
e = 0;
242+
}
243+
}
244+
}
245+
}
246+
}

0 commit comments

Comments
 (0)