Pivot Operator

Author: benjchristensen

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

@@ -12,7 +12,7 @@
  */
 package rx;
 
-import static rx.functions.Functions.*;
+import static rx.functions.Functions.alwaysFalse;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -115,6 +115,7 @@
 import rx.operators.OperatorOnErrorFlatMap;
 import rx.operators.OperatorOnErrorResumeNextViaFunction;
 import rx.operators.OperatorParallel;
+import rx.operators.OperatorPivot;
 import rx.operators.OperatorRepeat;
 import rx.operators.OperatorRetry;
 import rx.operators.OperatorScan;
@@ -1640,8 +1641,18 @@ public final static Observable<Long> interval(long interval, TimeUnit unit, Sche
      * @return an Observable that emits {@code value} as a single item and then completes
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#wiki-just">RxJava Wiki: just()</a>
      */
-    public final static <T> Observable<T> just(T value) {
-        return from(Arrays.asList(value));
+    public final static <T> Observable<T> just(final T value) {
+        return Observable.create(new OnSubscribe<T>() {
+
+            @Override
+            public void call(Subscriber<? super T> s) {
+                if (!s.isUnsubscribed()) {
+                    s.onNext(value);
+                    s.onCompleted();
+                }
+            }
+
+        });
     }
 
     /**
@@ -1664,7 +1675,7 @@ public final static <T> Observable<T> just(T value) {
      */
     @Deprecated
     public final static <T> Observable<T> just(T value, Scheduler scheduler) {
-        return from(Arrays.asList((value)), scheduler);
+        return just(value).subscribeOn(scheduler);
     }
 
     /**
@@ -2410,7 +2421,7 @@ public final static <T extends Comparable<? super T>> Observable<T> min(Observab
      * @return an Observable that emits a single item: the source Observable
      */
     public final Observable<Observable<T>> nest() {
-        return from(this);
+        return just(this);
     }
 
     /**
@@ -2477,6 +2488,16 @@ public final static <T> Observable<Observable<T>> parallelMerge(Observable<Obser
         return OperationParallelMerge.parallelMerge(source, parallelObservables, scheduler);
     }
 
+    /**
+     * Pivot GroupedObservable streams without serializing/synchronizing to a single stream first. 
+     * 
+     * @param groups
+     * @return
+     */
+    public static final <K1, K2, T> Observable<GroupedObservable<K2, GroupedObservable<K1, T>>> pivot(Observable<GroupedObservable<K1, GroupedObservable<K2, T>>> groups) {
+        return groups.lift(new OperatorPivot<K1, K2, T>());        
+    }
+    
     /**
      * Returns an Observable that emits a sequence of Integers within a specified range.
      * <p>

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -16,6 +16,7 @@
 package rx.observables;
 
 import rx.Observable;
+import rx.Subscriber;
 import rx.functions.Func1;
 
 /**
@@ -31,6 +32,16 @@
 public class GroupedObservable<K, T> extends Observable<T> {
     private final K key;
 
+    public static <K, T> GroupedObservable<K, T> from(K key, final Observable<T> o) {
+        return new GroupedObservable<K, T>(key, new OnSubscribe<T>() {
+
+            @Override
+            public void call(Subscriber<? super T> s) {
+                o.subscribe(s);
+            }
+        });
+    }
+
     public GroupedObservable(K key, OnSubscribe<T> onSubscribe) {
         super(onSubscribe);
         this.key = key;

rxjava-core/src/main/java/rx/observers/SerializedObserver.java

@@ -2,9 +2,6 @@
 
 import java.util.ArrayList;
 
-import javax.management.NotificationListener;
-
-import rx.Notification;
 import rx.Observer;
 import rx.operators.NotificationLite;
 
@@ -54,6 +51,7 @@ public void onCompleted() {
                 queue.add(on.completed());
             }
         }
+
         if (canEmit) {
             // we won the right to emit
             try {