OperatorConcat

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

@@ -50,7 +50,6 @@
 import rx.operators.OnSubscribeRange;
 import rx.operators.OperationBuffer;
 import rx.operators.OperationCombineLatest;
-import rx.operators.OperationConcat;
 import rx.operators.OperationDebounce;
 import rx.operators.OperationDefaultIfEmpty;
 import rx.operators.OperationDefer;
@@ -95,6 +94,7 @@
 import rx.operators.OperatorAsObservable;
 import rx.operators.OperatorCache;
 import rx.operators.OperatorCast;
+import rx.operators.OperatorConcat;
 import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorElementAt;
 import rx.operators.OperatorFilter;
@@ -740,7 +740,7 @@ public final static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> combin
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>
      */
     public final static <T> Observable<T> concat(Observable<? extends Observable<? extends T>> observables) {
-        return create(OperationConcat.concat(observables));
+        return observables.lift(new OperatorConcat<T>());
     }
 
     /**
@@ -761,7 +761,7 @@ public final static <T> Observable<T> concat(Observable<? extends Observable<? e
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2) {
-        return create(OperationConcat.concat(t1, t2));
+        return concat(from(t1, t2));
     }
 
     /**
@@ -784,7 +784,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3) {
-        return create(OperationConcat.concat(t1, t2, t3));
+        return concat(from(t1, t2, t3));
     }
 
     /**
@@ -809,7 +809,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4) {
-        return create(OperationConcat.concat(t1, t2, t3, t4));
+        return concat(from(t1, t2, t3, t4));
     }
 
     /**
@@ -836,7 +836,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     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) {
-        return create(OperationConcat.concat(t1, t2, t3, t4, t5));
+        return concat(from(t1, t2, t3, t4, t5));
     }
 
     /**
@@ -865,7 +865,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     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) {
-        return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6));
+        return concat(from(t1, t2, t3, t4, t5, t6));
     }
 
     /**
@@ -896,7 +896,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     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) {
-        return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7));
+        return concat(from(t1, t2, t3, t4, t5, t6, t7));
     }
 
     /**
@@ -929,7 +929,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     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) {
-        return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7, t8));
+        return concat(from(t1, t2, t3, t4, t5, t6, t7, t8));
     }
 
     /**
@@ -964,7 +964,7 @@ public final static <T> Observable<T> concat(Observable<? extends T> t1, Observa
     @SuppressWarnings("unchecked")
     // suppress because the types are checked by the method signature before using a vararg
     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) {
-        return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7, t8, t9));
+        return concat(from(t1, t2, t3, t4, t5, t6, t7, t8, t9));
     }
 
     /**

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

@@ -0,0 +1,119 @@
+ /**
+  * Copyright 2014 Netflix, Inc.
+  *
+  * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+  * use this file except in compliance with the License. You may obtain a copy of
+  * the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+  * License for the specific language governing permissions and limitations under
+  * the License.
+  */
+package rx.operators;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import rx.Observable;
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.observers.SerializedSubscriber;
+import rx.subscriptions.SerialSubscription;
+import rx.subscriptions.Subscriptions;
+
+/**
+ * Returns an Observable that emits the items emitted by two or more Observables, one after the
+ * other.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/concat.png">
+ */
+public final class OperatorConcat<T> implements Operator<T, Observable<? extends T>> {
+    final NotificationLite<Observable<? extends T>> nl = NotificationLite.instance();
+    @Override
+    public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
+        final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
+        final SerialSubscription current = new SerialSubscription();
+        child.add(current);
+        return new ConcatSubscriber(s, current);
+    }
+
+    final class ConcatSubscriber extends Subscriber<Observable<? extends T>> {
+
+        private final Subscriber<T> s;
+        private final SerialSubscription current;
+        final ConcurrentLinkedQueue<Object> queue;
+        final AtomicInteger wip;
+
+        public ConcatSubscriber(Subscriber<T> s, SerialSubscription current) {
+            super(s);
+            this.s = s;
+            this.current = current;
+            this.queue = new ConcurrentLinkedQueue<Object>();
+            this.wip = new AtomicInteger();
+            add(Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    queue.clear();
+                }
+            }));
+        }
+
+        @Override
+        public void onNext(Observable<? extends T> t) {
+            queue.add(nl.next(t));
+            if (wip.getAndIncrement() == 0) {
+                subscribeNext();
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            s.onError(e);
+            unsubscribe();
+        }
+
+        @Override
+        public void onCompleted() {
+            queue.add(nl.completed());
+            if (wip.getAndIncrement() == 0) {
+                subscribeNext();
+            }
+        }
+
+        void subscribeNext() {
+            Object o = queue.poll();
+            if (nl.isCompleted(o)) {
+                s.onCompleted();
+            } else 
+            if (o != null) {
+                Observable<? extends T> obs = nl.getValue(o);
+                Subscriber<T> sourceSub = new Subscriber<T>() {
+
+                    @Override
+                    public void onNext(T t) {
+                        s.onNext(t);
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        ConcatSubscriber.this.onError(e);
+                    }
+
+                    @Override
+                    public void onCompleted() {
+                        if (wip.decrementAndGet() > 0) {
+                            subscribeNext();
+                        }
+                    }
+
+                };
+                current.set(sourceSub);
+                obs.unsafeSubscribe(sourceSub);
+            }
+        }
+    }
+}