Refactoring, renaming, and simplifying

Author: Aaron Tull

src/main/java/rx/dual/DualConversion.java

@@ -0,0 +1,5 @@
+package rx.dual;
+
+public interface DualConversion<O, T1, T2> {
+    public O convert(DualOnSubscribe<T1, T2> onSubscribe);
+}

src/main/java/rx/dual/DualExtendingOperator.java

@@ -4,6 +4,5 @@
 public interface DualObservable<T1, T2> {
     public void subscribe(DualSubscriber<T1, T2> subscriber);
     public <R1, R2> DualObservable<R1, R2> lift(final DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator);
-    public <R1, R2, O extends DualObservable<R1, R2>> O lift(DualObservableFactory<O, R1, R2, T1, T2> factory, DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator);
-    public <R1, R2, O extends DualObservable<R1, R2>> O extend(DualExtendingOperator<O, R1, R2, T1, T2> operator);
+    public <O> O extend(DualConversion<O, T1, T2> operator);
 }

src/main/java/rx/dual/DualObservable.java

@@ -0,0 +1,7 @@
+package rx.dual;
+
+import rx.Observable.OnSubscribe;
+
+public interface MonoToDualConversion<O extends DualObservable<R1, R2>, R1, R2, T> {
+    public O convert(OnSubscribe<T> onSubscribe);
+}

src/main/java/rx/dual/DualObservableFactory.java

@@ -3,7 +3,7 @@
 import rx.Subscriber;
 import rx.functions.Func1;
 
-public interface SingleToDualOperator<R1, R2, T> extends Func1<DualSubscriber<? super R1, ? super R2>, Subscriber<? super T>> {
+public interface MonoToDualOperator<R1, R2, T> extends Func1<DualSubscriber<? super R1, ? super R2>, Subscriber<? super T>> {
     @Override
     public Subscriber<? super T> call(DualSubscriber<? super R1, ? super R2> subscriber);
 }

src/main/java/rx/dual/MonoToDualConversion.java

@@ -0,0 +1,66 @@
+package rx.simple;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.functions.Func2;
+import rx.internal.operators.OperatorFilter;
+import rx.internal.operators.OperatorMap;
+import rx.internal.operators.OperatorScan;
+import rx.single.MonoConversion;
+import rx.single.MonoObservable;
+
+public class CoreObservable<T> extends SimpleMonoObservable<T> {
+    
+    private CoreObservable(OnSubscribe<T> onSubscribe) {
+        super(onSubscribe);
+    }
+
+    public static <T> CoreObservable<T> create(OnSubscribe<T> onSubscribe) {
+        return new CoreObservable<T>(onSubscribe);
+    }
+
+    @Override
+    public void subscribe(Subscriber<T> subscriber) {
+
+    }
+
+    @Override
+    public <R> CoreObservable<R> lift(Operator<? extends R, ? super T> operator) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    @Override
+    public <R, O> O extend(MonoConversion<O, T> operator) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    @Override
+    public <R> MonoObservable<? extends R> compose(Func1<? super MonoObservable<? super T>, ? extends MonoObservable<? extends R>> composition) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    public final <R> CoreObservable<R> map(Func1<? super T, ? extends R> func) {
+        return lift(new OperatorMap<T, R>(func));
+    }
+
+    public final CoreObservable<T> filter(Func1<? super T, Boolean> predicate) {
+        return lift(new OperatorFilter<T>(predicate));
+    }
+    
+    public final <R> CoreObservable<R> scan(R initialValue, Func2<R, ? super T, R> accumulator) {
+        return lift(new OperatorScan<R, T>(initialValue, accumulator));
+    }
+
+    public final <R> CoreObservable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
+//        return map(func).lift(OperatorMerge.<T>instance(false));
+//        return map(func).extend(new CoreObservableConversion<R, Observable<R>>(OperatorMerge.<T>instance(false)));
+        return null;
+    }
+
+}

src/main/java/rx/dual/SingleToDualOperator.java renamed to src/main/java/rx/dual/MonoToDualOperator.java

@@ -0,0 +1,50 @@
+package rx.simple;
+
+import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.exceptions.OnErrorNotImplementedException;
+import rx.single.MonoConversion;
+
+public class CoreObservableConversion<R, T> implements MonoConversion<CoreObservable<R>, T> {
+
+    private Operator<? extends R, ? super T> operator;
+
+    public CoreObservableConversion(Operator<? extends R, ? super T> operator) {
+        this.operator = operator;
+    }
+
+    @Override
+    public CoreObservable<R> convert(OnSubscribe<T> onSubscribe) {
+        return CoreObservable.create(wrapSubscriber(onSubscribe, operator));
+    }
+
+    protected OnSubscribe<R> wrapSubscriber(OnSubscribe<T> onSubscribe, Operator<? extends R, ? super T> operator) {
+        return (Subscriber<? super R> o) -> {
+            try {
+                Subscriber<? super T> st = operator.call(o);
+                try {
+                    // new Subscriber created and being subscribed with so 'onStart' it
+                    st.onStart();
+                    onSubscribe.call(st);
+                } catch (Throwable e) {
+                    // localized capture of errors rather than it skipping all operators 
+                    // and ending up in the try/catch of the subscribe method which then
+                    // prevents onErrorResumeNext and other similar approaches to error handling
+                    if (e instanceof OnErrorNotImplementedException) {
+                        throw e;
+                    }
+                    st.onError(e);
+                }
+            } catch (Throwable e) {
+                if (e instanceof OnErrorNotImplementedException) {
+                    throw e;
+                }
+                // if the lift function failed all we can do is pass the error to the final Subscriber
+                // as we don't have the operator available to us
+                o.onError(e);
+            }
+        };
+    }
+
+}

src/main/java/rx/dual/SingleToDualExtendingOperator.java

@@ -1,21 +1,21 @@
 package rx.simple;
 
-import rx.dual.DualExtendingOperator;
+import rx.dual.DualConversion;
 import rx.dual.DualOnSubscribe;
 import rx.dual.DualOperator;
 import rx.dual.DualSubscriber;
 import rx.exceptions.OnErrorNotImplementedException;
 
-public class SimpleDualExtendingOperator<R1, R2, T1, T2> implements DualExtendingOperator<SimpleDualObservable<R1, R2>, R1, R2, T1, T2> {
+public class SimpleDualConversion<R1, R2, T1, T2> implements DualConversion<SimpleDualObservable<R1, R2>, T1, T2> {
 
-    private DualOperator<R1, R2, T1, T2> operator;
+    private DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator;
 
-    public SimpleDualExtendingOperator(DualOperator<R1, R2, T1, T2> operator) {
+    public SimpleDualConversion(DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator) {
         this.operator = operator;
     }
 
     @Override
-    public SimpleDualObservable<R1, R2> compose(DualOnSubscribe<T1, T2> onSubscribe) {
+    public SimpleDualObservable<R1, R2> convert(DualOnSubscribe<T1, T2> onSubscribe) {
         return SimpleDualObservable.create(wrapSubscriber(onSubscribe));
     }
 

src/main/java/rx/dual/SingleToDualObservableFactory.java

@@ -1,8 +1,7 @@
 package rx.simple;
 
-import rx.dual.DualExtendingOperator;
+import rx.dual.DualConversion;
 import rx.dual.DualObservable;
-import rx.dual.DualObservableFactory;
 import rx.dual.DualOnSubscribe;
 import rx.dual.DualOperator;
 import rx.dual.DualSubscriber;
@@ -26,17 +25,12 @@ public void subscribe(DualSubscriber<T1, T2> subscriber) {
 
     @Override
     public <R1, R2> DualObservable<R1, R2> lift(DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator) {
-        return lift(new SimpleDualObservableFactory<R1, R2, T1, T2>(), operator);
+        return extend(new SimpleDualConversion<R1, R2, T1, T2>(operator));
     }
-
-    @Override
-    public <R1, R2, O extends DualObservable<R1, R2>> O lift(DualObservableFactory<O, R1, R2, T1, T2> factory, DualOperator<? extends R1, ? extends R2, ? super T1, ? super T2> operator) {
-        return factory.call(operator, onSubscribe);
-    }
-    
+   
     @Override
-    public <R1, R2, O extends DualObservable<R1, R2>> O extend(DualExtendingOperator<O, R1, R2, T1, T2> operator) {
-        return operator.compose(onSubscribe);
+    public <O> O extend(DualConversion<O, T1, T2> operator) {
+        return operator.convert(onSubscribe);
     }
 
 }

src/main/java/rx/simple/CoreObservable.java

@@ -4,19 +4,19 @@
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.exceptions.OnErrorNotImplementedException;
-import rx.single.SingleExtendingOperator;
+import rx.single.MonoConversion;
 
-public class SimpleExtendingOperator<R, T> implements SingleExtendingOperator<SimpleSingleObservable<R>, R, T> {
+public class SimpleMonoConversion<R, T> implements MonoConversion<SimpleMonoObservable<R>, T> {
 
-    private Operator<R, T> operator;
+    private Operator<? extends R, ? super T> operator;
 
-    public SimpleExtendingOperator(Operator<R, T> operator) {
+    public SimpleMonoConversion(Operator<? extends R, ? super T> operator) {
         this.operator = operator;
     }
 
     @Override
-    public SimpleSingleObservable<R> compose(OnSubscribe<T> onSubscribe) {
-        return SimpleSingleObservable.create(wrapSubscriber(onSubscribe));
+    public SimpleMonoObservable<R> convert(OnSubscribe<T> onSubscribe) {
+        return SimpleMonoObservable.create(wrapSubscriber(onSubscribe));
     }
 
     private OnSubscribe<R> wrapSubscriber(OnSubscribe<T> onSubscribe) {

src/main/java/rx/simple/CoreObservableConversion.java

@@ -0,0 +1,41 @@
+package rx.simple;
+
+import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.single.MonoConversion;
+import rx.single.MonoObservable;
+
+public class SimpleMonoObservable<T> implements MonoObservable<T> {
+
+    private OnSubscribe<T> onSubscribe;
+    
+    public static <T> SimpleMonoObservable<T> create(OnSubscribe<T> onSubscribe) {
+        return new SimpleMonoObservable<T>(onSubscribe);
+    }
+
+    protected SimpleMonoObservable(OnSubscribe<T> onSubscribe) {
+        this.onSubscribe = onSubscribe;
+    }
+
+    @Override
+    public void subscribe(Subscriber<T> subscriber) {
+        onSubscribe.call(subscriber);
+    }
+
+    @Override
+    public <R> SimpleMonoObservable<R> lift(Operator<? extends R, ? super T> operator) {
+        return extend(new SimpleMonoConversion<R, T>(operator));
+    }
+    
+    public <R, O> O extend(MonoConversion<O, T> operator) {
+        return operator.convert(onSubscribe);
+    }
+
+    @Override
+    public <R> MonoObservable<? extends R> compose(Func1<? super MonoObservable<? super T>, ? extends MonoObservable<? extends R>> transformer) {
+        return transformer.call(this);
+    }
+    
+}

src/main/java/rx/simple/SimpleDualExtendingOperator.java renamed to src/main/java/rx/simple/SimpleDualConversion.java

@@ -4,19 +4,19 @@
 import rx.Subscriber;
 import rx.dual.DualOnSubscribe;
 import rx.dual.DualSubscriber;
-import rx.dual.SingleToDualExtendingOperator;
-import rx.dual.SingleToDualOperator;
+import rx.dual.MonoToDualOperator;
 import rx.exceptions.OnErrorNotImplementedException;
+import rx.single.MonoConversion;
 
-public class SimpleSingleToDualExtendingOperator<R1, R2, T> implements SingleToDualExtendingOperator<SimpleDualObservable<R1, R2>, R1, R2, T> {
-    private SingleToDualOperator<R1, R2, T> operator;
+public class SimpleMonoToDualConversion<R1, R2, T> implements MonoConversion<SimpleDualObservable<R1, R2>, T> {
+    private MonoToDualOperator<R1, R2, T> operator;
 
-    public SimpleSingleToDualExtendingOperator(SingleToDualOperator<R1, R2, T> operator) {
+    public SimpleMonoToDualConversion(MonoToDualOperator<R1, R2, T> operator) {
         this.operator = operator;
     }
 
     @Override
-    public SimpleDualObservable<R1, R2> compose(OnSubscribe<T> onSubscribe) {
+    public SimpleDualObservable<R1, R2> convert(OnSubscribe<T> onSubscribe) {
         return SimpleDualObservable.create(wrapSubscriber(onSubscribe));
     }