Refactoring and renaming. Removed Dual observable.

Author: Aaron Tull

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

@@ -8,41 +8,31 @@
 import rx.functions.Func2;
 import rx.internal.operators.OperatorFilter;
 import rx.internal.operators.OperatorMap;
+import rx.internal.operators.OperatorMerge;
 import rx.internal.operators.OperatorScan;
 import rx.single.MonoConversion;
-import rx.single.MonoObservable;
 
-public class CoreObservable<T> extends SimpleMonoObservable<T> {
+public class CoreObservable<T> {
+    protected OnSubscribe<T> onSubscribe;
 
     private CoreObservable(OnSubscribe<T> onSubscribe) {
-        super(onSubscribe);
+        this.onSubscribe = onSubscribe;
     }
 
     public static <T> CoreObservable<T> create(OnSubscribe<T> onSubscribe) {
         return new CoreObservable<T>(onSubscribe);
     }
 
-    @Override
     public void subscribe(Subscriber<T> subscriber) {
-
+        onSubscribe.call(subscriber);
     }
 
-    @Override
     public <R> CoreObservable<R> lift(Operator<? extends R, ? super T> operator) {
-        // TODO Auto-generated method stub
-        return null;
+        return extend(new CoreObservableConversion<R, T>(operator));
     }
 
-    @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;
+        return operator.convert(onSubscribe);
     }
 
     public final <R> CoreObservable<R> map(Func1<? super T, ? extends R> func) {
@@ -59,7 +49,12 @@ public final <R> CoreObservable<R> scan(R initialValue, Func2<R, ? super T, R> a
 
     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 map(func).extend(new MonoConversion<CoreObservable<R>, R>(){
+//
+//            @Override
+//            public CoreObservable<R> convert(OnSubscribe<R> onSubscribe) {
+//            }});
         return null;
     }
 

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

@@ -2,6 +2,7 @@
 
 import rx.Observable.OnSubscribe;
 import rx.Observable.Operator;
+import rx.Observable;
 import rx.Subscriber;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.single.MonoConversion;

src/main/java/rx/dual/DualObserver.java

@@ -5,11 +5,10 @@
 import rx.Subscriber;
 import rx.functions.Func1;
 import rx.single.MonoConversion;
-import rx.single.MonoObservable;
 
-public class SimpleMonoObservable<T> implements MonoObservable<T> {
+public class SimpleMonoObservable<T> {
 
-    private OnSubscribe<T> onSubscribe;
+    protected OnSubscribe<T> onSubscribe;
 
     public static <T> SimpleMonoObservable<T> create(OnSubscribe<T> onSubscribe) {
         return new SimpleMonoObservable<T>(onSubscribe);
@@ -19,12 +18,10 @@ 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));
     }
@@ -33,8 +30,7 @@ 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) {
+    public <R> SimpleMonoObservable<? extends R> compose(Func1<SimpleMonoObservable<? super T>, SimpleMonoObservable<? extends R>> transformer) {
         return transformer.call(this);
     }
 

src/main/java/rx/dual/DualOnSubscribe.java

@@ -1,21 +1,20 @@
 package rx.simple.operator;
 
+import rx.Observable.Operator;
 import rx.Subscriber;
-import rx.dual.DualSubscriber;
-import rx.dual.MonoToDualOperator;
 import rx.functions.Func1;
-import rx.simple.SimpleDualObservable;
-import rx.simple.SimpleMonoToDualConversion;
+import rx.simple.CoreObservable;
+import rx.simple.CoreObservableConversion;
 import rx.single.MonoConversion;
 
 public class ConversionOperators {
 
-    public static <T1, R1> MonoConversion<SimpleDualObservable<T1, R1>, T1> generate(Func1<? super T1, ? extends R1> generatorFunc) {
-        return new SimpleMonoToDualConversion<T1, R1, T1>(new MonoToDualOperator<T1, R1, T1>() {
+    public static <T, R> MonoConversion<CoreObservable<R>, T> mapToCore(Func1<? super T, ? extends R> mapFunc) {
+        return new CoreObservableConversion<R, T>(new Operator<R, T>() {
 
             @Override
-            public Subscriber<? super T1> call(DualSubscriber<? super T1, ? super R1> subscriber) {
-                return new Subscriber<T1>() {
+            public Subscriber<? super T> call(Subscriber<? super R> subscriber) {
+                return new Subscriber<T>() {
                     @Override
                     public void onCompleted() {
                         subscriber.onCompleted();
@@ -27,10 +26,10 @@ public void onError(Throwable e) {
                     }
 
                     @Override
-                    public void onNext(T1 t0) {
+                    public void onNext(T t) {
                         try {
-                            R1 t1 = generatorFunc.call(t0);
-                            subscriber.onNext(t0, t1);
+                            R r = mapFunc.call(t);
+                            subscriber.onNext(r);
                         } catch (Throwable e) {
                             subscriber.onError(e);
                         }