Added dual map extending operator

Author: Aaron Tull

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

@@ -1,7 +1,10 @@
 package rx.dual;
 
-import rx.Observer;
 
-public interface DualObserver<T1, T2> extends Observer<T1> {
+public interface DualObserver<T1, T2> {
     public void onNext(T1 t1, T2 t2);
+
+    public void onCompleted();
+
+    public void onError(Throwable e);
 }

src/main/java/rx/dual/DualSubscriber.java

@@ -1,8 +1,9 @@
 package rx.dual;
 
-import rx.Subscriber;
+import rx.Subscription;
 
-public abstract class DualSubscriber<T1, T2> extends Subscriber<T1> implements DualObserver<T1, T2> {
+
+public abstract class DualSubscriber<T1, T2> implements DualObserver<T1, T2>, Subscription {
 
     @Override
     public abstract void onCompleted();
@@ -11,9 +12,20 @@ public abstract class DualSubscriber<T1, T2> extends Subscriber<T1> implements D
     public abstract void onError(Throwable e);
 
     @Override
-    public void onNext(T1 t) {}
+    public abstract void onNext(T1 t1, T2 t2);
 
+    public void onStart() {
+        
+    }
+    
     @Override
-    public abstract void onNext(T1 t1, T2 t2);
+    public void unsubscribe() {
+        
+    }
+
+    @Override
+    public boolean isUnsubscribed() {
+        return false;
+    }
 
 }

src/main/java/rx/simple/SimpleDualExtendingOperator.java

@@ -10,7 +10,7 @@ public class SimpleDualExtendingOperator<R1, R2, T1, T2> implements DualExtendin
 
     private DualOperator<R1, R2, T1, T2> operator;
 
-    SimpleDualExtendingOperator(DualOperator<R1, R2, T1, T2> operator) {
+    public SimpleDualExtendingOperator(DualOperator<R1, R2, T1, T2> operator) {
         this.operator = operator;
     }
 

src/main/java/rx/simple/operator/DualOperators.java

@@ -0,0 +1,36 @@
+package rx.simple.operator;
+
+import rx.dual.DualExtendingOperator;
+import rx.dual.DualOperator;
+import rx.dual.DualSubscriber;
+import rx.functions.Func2;
+import rx.simple.SimpleDualExtendingOperator;
+import rx.simple.SimpleDualObservable;
+
+public class DualOperators {
+
+    public static <T1, T2, R1> DualExtendingOperator<SimpleDualObservable<R1, T2>, R1, T2, T1, T2> map1(Func2<? super T1, ? super T2, ? extends R1> mapFunc) {
+        return new SimpleDualExtendingOperator<R1, T2, T1, T2>(new DualOperator<R1, T2, T1, T2>() {
+
+            @Override
+            public DualSubscriber<? super T1, ? super T2> call(DualSubscriber<? super R1, ? super T2> subscriber) {
+                return new DualSubscriber<T1, T2>() {
+
+                    @Override
+                    public void onCompleted() {
+                        subscriber.onCompleted();
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        subscriber.onError(e);
+                    }
+
+                    @Override
+                    public void onNext(T1 t1, T2 t2) {
+                        subscriber.onNext(mapFunc.call(t1, t2), t2);
+                    }};
+            }});
+    }
+
+}

src/test/java/rx/simple/SimpleSingleObservableTest.java

@@ -3,6 +3,7 @@
 import org.junit.Test;
 
 import static rx.simple.operator.CoreOperators.*;
+import static rx.simple.operator.DualOperators.*;
 import static rx.simple.operator.ConversionOperators.*;
 import rx.Subscriber;
 import rx.dual.DualSubscriber;
@@ -51,7 +52,12 @@ public void testConversion() {
                 sub.onNext(new Integer(i));
             sub.onCompleted();
         })
-        .extend(generate(i -> i.toString()))
+        .extend(
+                generate(i -> i.toString())
+        )
+        .extend(
+                map1((Integer i, String s) -> i * s.length())
+        )
         .subscribe(new DualSubscriber<Integer, String>() {
             @Override
             public void onNext(Integer t1, String t2) {