Skip to content

Commit b9fe278

Browse files
benjchristensenbenjchristensen
committed
Merge pull request #869 from benjchristensen/subscribeOn+groupBy
subscribeOn + groupBy
2 parents 1e6224d + 6863f57 commit b9fe278

File tree

8 files changed

+581
-84
lines changed

8 files changed

+581
-84
lines changed

‎rxjava-core/src/main/java/rx/Observable.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7102,11 +7102,12 @@ public final Subscription subscribe(Subscriber<? super T> observer, Scheduler sc
71027102
* @return the source Observable modified so that its subscriptions and unsubscriptions happen on the
71037103
* specified {@link Scheduler}
71047104
* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-subscribeon">RxJava Wiki: subscribeOn()</a>
7105+
* @see #subscribeOn(rx.Scheduler, int)
71057106
*/
71067107
public final Observable<T> subscribeOn(Scheduler scheduler) {
71077108
return nest().lift(new OperatorSubscribeOn<T>(scheduler));
71087109
}
7109-
7110+
71107111
/**
71117112
* Returns an Observable that extracts a Double from each of the items emitted by the source Observable via
71127113
* a function you specify, and then emits the sum of these Doubles.

‎rxjava-core/src/main/java/rx/observers/TestSubscriber.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,14 @@ public void awaitTerminalEvent(long timeout, TimeUnit unit) {
101101
}
102102
}
103103

104+
public void awaitTerminalEventAndUnsubscribeOnTimeout(long timeout, TimeUnit unit) {
105+
try {
106+
awaitTerminalEvent(timeout, unit);
107+
} catch (RuntimeException e) {
108+
unsubscribe();
109+
}
110+
}
111+
104112
public Thread getLastSeenThread() {
105113
return lastSeenThread;
106114
}

‎rxjava-core/src/main/java/rx/operators/BufferUntilSubscriber.java‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
/**
2+
* Copyright 2014 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package rx.operators;
17+
18+
import java.util.LinkedList;
19+
import java.util.Queue;
20+
import rx.Subscriber;
21+
import rx.subscriptions.CompositeSubscription;
22+
23+
/**
24+
* Buffers the incoming events until notified, then replays the
25+
* buffered events and continues as a simple pass-through subscriber.
26+
* @param <T> the streamed value type
27+
*/
28+
public class BufferUntilSubscriber<T> extends Subscriber<T> {
29+
/** The actual subscriber. */
30+
private final Subscriber<? super T> actual;
31+
/** Indicate the pass-through mode. */
32+
private volatile boolean passthroughMode;
33+
/** Protect mode transition. */
34+
private final Object gate = new Object();
35+
/** The buffered items. */
36+
private final Queue<Object> queue = new LinkedList<Object>();
37+
/** The queue capacity. */
38+
private final int capacity;
39+
/** Null sentinel (in case queue type is changed). */
40+
private static final Object NULL_SENTINEL = new Object();
41+
/** Complete sentinel. */
42+
private static final Object COMPLETE_SENTINEL = new Object();
43+
/**
44+
* Container for an onError event.
45+
*/
46+
private static final class ErrorSentinel {
47+
final Throwable t;
48+
49+
public ErrorSentinel(Throwable t) {
50+
this.t = t;
51+
}
52+
53+
}
54+
/**
55+
* Constructor that wraps the actual subscriber and shares its subscription.
56+
* @param capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
57+
* @param actual
58+
*/
59+
public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual) {
60+
super(actual);
61+
this.actual = actual;
62+
this.capacity = capacity;
63+
}
64+
/**
65+
* Constructor that wraps the actual subscriber and uses the given composite
66+
* subscription.
67+
* @param capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
68+
* @param actual
69+
* @param cs
70+
*/
71+
public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual, CompositeSubscription cs) {
72+
super(cs);
73+
this.actual = actual;
74+
this.capacity = capacity;
75+
}
76+
77+
/**
78+
* Call this method to replay the buffered events and continue as a pass-through subscriber.
79+
* If already in pass-through mode, this method is a no-op.
80+
*/
81+
public void enterPassthroughMode() {
82+
if (!passthroughMode) {
83+
synchronized (gate) {
84+
if (!passthroughMode) {
85+
while (!queue.isEmpty()) {
86+
Object o = queue.poll();
87+
if (!actual.isUnsubscribed()) {
88+
if (o == NULL_SENTINEL) {
89+
actual.onNext(null);
90+
} else
91+
if (o == COMPLETE_SENTINEL) {
92+
actual.onCompleted();
93+
} else
94+
if (o instanceof ErrorSentinel) {
95+
actual.onError(((ErrorSentinel)o).t);
96+
} else
97+
if (o != null) {
98+
@SuppressWarnings("unchecked")
99+
T v = (T)o;
100+
actual.onNext(v);
101+
} else {
102+
throw new NullPointerException();
103+
}
104+
}
105+
}
106+
passthroughMode = true;
107+
gate.notifyAll();
108+
}
109+
}
110+
}
111+
}
112+
@Override
113+
public void onNext(T t) {
114+
if (!passthroughMode) {
115+
synchronized (gate) {
116+
if (!passthroughMode) {
117+
if (capacity < 0 || queue.size() < capacity) {
118+
queue.offer(t != null ? t : NULL_SENTINEL);
119+
return;
120+
}
121+
try {
122+
while (!passthroughMode) {
123+
gate.wait();
124+
}
125+
if (actual.isUnsubscribed()) {
126+
return;
127+
}
128+
} catch (InterruptedException ex) {
129+
Thread.currentThread().interrupt();
130+
actual.onError(ex);
131+
return;
132+
}
133+
}
134+
}
135+
}
136+
actual.onNext(t);
137+
}
138+
139+
@Override
140+
public void onError(Throwable e) {
141+
if (!passthroughMode) {
142+
synchronized (gate) {
143+
if (!passthroughMode) {
144+
if (capacity < 0 || queue.size() < capacity) {
145+
queue.offer(new ErrorSentinel(e));
146+
return;
147+
}
148+
try {
149+
while (!passthroughMode) {
150+
gate.wait();
151+
}
152+
if (actual.isUnsubscribed()) {
153+
return;
154+
}
155+
} catch (InterruptedException ex) {
156+
Thread.currentThread().interrupt();
157+
actual.onError(ex);
158+
return;
159+
}
160+
}
161+
}
162+
}
163+
actual.onError(e);
164+
}
165+
166+
@Override
167+
public void onCompleted() {
168+
if (!passthroughMode) {
169+
synchronized (gate) {
170+
if (!passthroughMode) {
171+
if (capacity < 0 || queue.size() < capacity) {
172+
queue.offer(COMPLETE_SENTINEL);
173+
return;
174+
}
175+
try {
176+
while (!passthroughMode) {
177+
gate.wait();
178+
}
179+
if (actual.isUnsubscribed()) {
180+
return;
181+
}
182+
} catch (InterruptedException ex) {
183+
Thread.currentThread().interrupt();
184+
actual.onError(ex);
185+
return;
186+
}
187+
}
188+
}
189+
}
190+
actual.onCompleted();
191+
}
192+
193+
}

‎rxjava-core/src/main/java/rx/operators/OperatorParallel.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,9 @@ public Integer call(T t) {
5656

5757
@Override
5858
public Observable<R> call(GroupedObservable<Integer, T> g) {
59+
// Must use observeOn not subscribeOn because we have a single source behind groupBy.
60+
// The origin is already subscribed to, we are moving each group on to a new thread
61+
// but the origin itself can only be on a single thread.
5962
return f.call(g.observeOn(scheduler));
6063
}
6164
});

‎rxjava-core/src/main/java/rx/operators/OperatorSubscribeOn.java‎

Lines changed: 57 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,27 +20,54 @@
2020
import rx.Scheduler;
2121
import rx.Scheduler.Inner;
2222
import rx.Subscriber;
23-
import rx.subscriptions.CompositeSubscription;
24-
import rx.subscriptions.Subscriptions;
25-
import rx.util.functions.Action0;
2623
import rx.util.functions.Action1;
2724

2825
/**
29-
* Asynchronously subscribes and unsubscribes Observers on the specified Scheduler.
26+
* Subscribes and unsubscribes Observers on the specified Scheduler.
3027
* <p>
28+
* Will occur asynchronously except when subscribing to `GroupedObservable`, `PublishSubject` and possibly other "hot" Observables
29+
* in which case it will subscribe synchronously and buffer/block onNext calls until the subscribe has occurred.
30+
* <p>
31+
* See https://github.com/Netflix/RxJava/issues/844 for more information on the "time gap" issue that the synchronous
32+
* subscribe is solving.
33+
*
3134
* <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/subscribeOn.png">
3235
*/
3336
public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {
3437

3538
private final Scheduler scheduler;
39+
/**
40+
* Indicate that events fired between the original subscription time and
41+
* the actual subscription time should not get lost.
42+
*/
43+
private final boolean dontLoseEvents;
44+
/** The buffer size to avoid flooding. Negative value indicates an unbounded buffer. */
45+
private final int bufferSize;
3646

3747
public OperatorSubscribeOn(Scheduler scheduler) {
3848
this.scheduler = scheduler;
49+
this.dontLoseEvents = false;
50+
this.bufferSize = -1;
51+
}
52+
53+
/**
54+
* Construct a SubscribeOn operator.
55+
*
56+
* @param scheduler
57+
* the target scheduler
58+
* @param bufferSize
59+
* if dontLoseEvents == true, this indicates the buffer size. Filling the buffer will
60+
* block the source. -1 indicates an unbounded buffer
61+
*/
62+
public OperatorSubscribeOn(Scheduler scheduler, int bufferSize) {
63+
this.scheduler = scheduler;
64+
this.dontLoseEvents = true;
65+
this.bufferSize = bufferSize;
3966
}
4067

4168
@Override
4269
public Subscriber<? super Observable<T>> call(final Subscriber<? super T> subscriber) {
43-
return new Subscriber<Observable<T>>() {
70+
return new Subscriber<Observable<T>>(subscriber) {
4471

4572
@Override
4673
public void onCompleted() {
@@ -52,48 +79,33 @@ public void onError(Throwable e) {
5279
subscriber.onError(e);
5380
}
5481

82+
boolean checkNeedBuffer(Observable<?> o) {
83+
return dontLoseEvents;
84+
}
85+
5586
@Override
5687
public void onNext(final Observable<T> o) {
57-
scheduler.schedule(new Action1<Inner>() {
58-
59-
@Override
60-
public void call(final Inner inner) {
61-
final CompositeSubscription cs = new CompositeSubscription();
62-
subscriber.add(Subscriptions.create(new Action0() {
63-
64-
@Override
65-
public void call() {
66-
inner.schedule(new Action1<Inner>() {
67-
68-
@Override
69-
public void call(final Inner inner) {
70-
cs.unsubscribe();
71-
}
72-
73-
});
74-
}
75-
76-
}));
77-
cs.add(subscriber);
78-
o.subscribe(new Subscriber<T>(cs) {
79-
80-
@Override
81-
public void onCompleted() {
82-
subscriber.onCompleted();
83-
}
84-
85-
@Override
86-
public void onError(Throwable e) {
87-
subscriber.onError(e);
88-
}
89-
90-
@Override
91-
public void onNext(T t) {
92-
subscriber.onNext(t);
93-
}
94-
});
95-
}
96-
});
88+
if (checkNeedBuffer(o)) {
89+
// use buffering (possibly blocking) for a possibly synchronous subscribe
90+
final BufferUntilSubscriber<T> bus = new BufferUntilSubscriber<T>(bufferSize, subscriber);
91+
o.subscribe(bus);
92+
subscriber.add(scheduler.schedule(new Action1<Inner>() {
93+
@Override
94+
public void call(final Inner inner) {
95+
bus.enterPassthroughMode();
96+
}
97+
}));
98+
return;
99+
} else {
100+
// no buffering (async subscribe)
101+
subscriber.add(scheduler.schedule(new Action1<Inner>() {
102+
103+
@Override
104+
public void call(final Inner inner) {
105+
o.subscribe(subscriber);
106+
}
107+
}));
108+
}
97109
}
98110

99111
};

0 commit comments

Comments
 (0)