Skip to content

Commit 923ea73

Browse files
benjchristensenbenjchristensen
committed
Merge pull request #1074 from akarnokd/OperatorConcat
OperatorConcat
2 parents 1de53a2 + b50a664 commit 923ea73

File tree

4 files changed

+207
-204
lines changed

4 files changed

+207
-204
lines changed

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,6 @@
4949
import rx.operators.OnSubscribeFromIterable;
5050
import rx.operators.OnSubscribeRange;
5151
import rx.operators.OperationCombineLatest;
52-
import rx.operators.OperationConcat;
5352
import rx.operators.OperationDebounce;
5453
import rx.operators.OperationDefaultIfEmpty;
5554
import rx.operators.OperationDefer;
@@ -98,6 +97,7 @@
9897
import rx.operators.OperatorBufferWithTime;
9998
import rx.operators.OperatorCache;
10099
import rx.operators.OperatorCast;
100+
import rx.operators.OperatorConcat;
101101
import rx.operators.OperatorDoOnEach;
102102
import rx.operators.OperatorElementAt;
103103
import rx.operators.OperatorFilter;
@@ -743,7 +743,7 @@ public final static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> combin
743743
* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>
744744
*/
745745
public final static <T> Observable<T> concat(Observable<? extends Observable<? extends T>> observables) {
746-
return create(OperationConcat.concat(observables));
746+
return observables.lift(new OperatorConcat<T>());
747747
}
748748

749749
/**
@@ -764,7 +764,7 @@ public final static <T> Observable<T> concat(Observable<? extends Observable<? e
764764
@SuppressWarnings("unchecked")
765765
// suppress because the types are checked by the method signature before using a vararg
766766
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2) {
767-
return create(OperationConcat.concat(t1, t2));
767+
return concat(from(t1, t2));
768768
}
769769

770770
/**
@@ -787,7 +787,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
787787
@SuppressWarnings("unchecked")
788788
// suppress because the types are checked by the method signature before using a vararg
789789
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3) {
790-
return create(OperationConcat.concat(t1, t2, t3));
790+
return concat(from(t1, t2, t3));
791791
}
792792

793793
/**
@@ -812,7 +812,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
812812
@SuppressWarnings("unchecked")
813813
// suppress because the types are checked by the method signature before using a vararg
814814
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4) {
815-
return create(OperationConcat.concat(t1, t2, t3, t4));
815+
return concat(from(t1, t2, t3, t4));
816816
}
817817

818818
/**
@@ -839,7 +839,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
839839
@SuppressWarnings("unchecked")
840840
// suppress because the types are checked by the method signature before using a vararg
841841
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5) {
842-
return create(OperationConcat.concat(t1, t2, t3, t4, t5));
842+
return concat(from(t1, t2, t3, t4, t5));
843843
}
844844

845845
/**
@@ -868,7 +868,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
868868
@SuppressWarnings("unchecked")
869869
// suppress because the types are checked by the method signature before using a vararg
870870
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6) {
871-
return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6));
871+
return concat(from(t1, t2, t3, t4, t5, t6));
872872
}
873873

874874
/**
@@ -899,7 +899,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
899899
@SuppressWarnings("unchecked")
900900
// suppress because the types are checked by the method signature before using a vararg
901901
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7) {
902-
return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7));
902+
return concat(from(t1, t2, t3, t4, t5, t6, t7));
903903
}
904904

905905
/**
@@ -932,7 +932,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
932932
@SuppressWarnings("unchecked")
933933
// suppress because the types are checked by the method signature before using a vararg
934934
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8) {
935-
return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7, t8));
935+
return concat(from(t1, t2, t3, t4, t5, t6, t7, t8));
936936
}
937937

938938
/**
@@ -967,7 +967,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
967967
@SuppressWarnings("unchecked")
968968
// suppress because the types are checked by the method signature before using a vararg
969969
public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8, Observable<? extends T> t9) {
970-
return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7, t8, t9));
970+
return concat(from(t1, t2, t3, t4, t5, t6, t7, t8, t9));
971971
}
972972

973973
/**

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

Lines changed: 0 additions & 175 deletions
This file was deleted.

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

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
/**
2+
* Copyright 2014 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* 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, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
package rx.operators;
17+
18+
import java.util.concurrent.ConcurrentLinkedQueue;
19+
import java.util.concurrent.atomic.AtomicInteger;
20+
import rx.Observable;
21+
import rx.Observable.Operator;
22+
import rx.Subscriber;
23+
import rx.functions.Action0;
24+
import rx.observers.SerializedSubscriber;
25+
import rx.subscriptions.SerialSubscription;
26+
import rx.subscriptions.Subscriptions;
27+
28+
/**
29+
* Returns an Observable that emits the items emitted by two or more Observables, one after the
30+
* other.
31+
* <p>
32+
* <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/concat.png">
33+
*/
34+
public final class OperatorConcat<T> implements Operator<T, Observable<? extends T>> {
35+
final NotificationLite<Observable<? extends T>> nl = NotificationLite.instance();
36+
@Override
37+
public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
38+
final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
39+
final SerialSubscription current = new SerialSubscription();
40+
child.add(current);
41+
return new ConcatSubscriber(s, current);
42+
}
43+
44+
final class ConcatSubscriber extends Subscriber<Observable<? extends T>> {
45+
46+
private final Subscriber<T> s;
47+
private final SerialSubscription current;
48+
final ConcurrentLinkedQueue<Object> queue;
49+
final AtomicInteger wip;
50+
51+
public ConcatSubscriber(Subscriber<T> s, SerialSubscription current) {
52+
super(s);
53+
this.s = s;
54+
this.current = current;
55+
this.queue = new ConcurrentLinkedQueue<Object>();
56+
this.wip = new AtomicInteger();
57+
add(Subscriptions.create(new Action0() {
58+
@Override
59+
public void call() {
60+
queue.clear();
61+
}
62+
}));
63+
}
64+
65+
@Override
66+
public void onNext(Observable<? extends T> t) {
67+
queue.add(nl.next(t));
68+
if (wip.getAndIncrement() == 0) {
69+
subscribeNext();
70+
}
71+
}
72+
73+
@Override
74+
public void onError(Throwable e) {
75+
s.onError(e);
76+
unsubscribe();
77+
}
78+
79+
@Override
80+
public void onCompleted() {
81+
queue.add(nl.completed());
82+
if (wip.getAndIncrement() == 0) {
83+
subscribeNext();
84+
}
85+
}
86+
87+
void subscribeNext() {
88+
Object o = queue.poll();
89+
if (nl.isCompleted(o)) {
90+
s.onCompleted();
91+
} else
92+
if (o != null) {
93+
Observable<? extends T> obs = nl.getValue(o);
94+
Subscriber<T> sourceSub = new Subscriber<T>() {
95+
96+
@Override
97+
public void onNext(T t) {
98+
s.onNext(t);
99+
}
100+
101+
@Override
102+
public void onError(Throwable e) {
103+
ConcatSubscriber.this.onError(e);
104+
}
105+
106+
@Override
107+
public void onCompleted() {
108+
if (wip.decrementAndGet() > 0) {
109+
subscribeNext();
110+
}
111+
}
112+
113+
};
114+
current.set(sourceSub);
115+
obs.unsafeSubscribe(sourceSub);
116+
}
117+
}
118+
}
119+
}

0 commit comments

Comments
 (0)