package com.github.davidmoten.rx;

import rx.Observable;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.functions.Func1;
import rx.subjects.PublishSubject;

/**
 * Converts an Operation (a function converting one Observable into another)
 * into an {@link Operator}.
 * 
 * @param <R>
 *            to type
 * @param <T>
 *            from type
 */
public class OperationToOperator<R, T> implements Operator<R, T> {

    public static <R, T> Operator<R, T> toOperator(Func1<Observable<T>, Observable<R>> operation) {
        return new OperationToOperator<R, T>(operation);
    }

    /**
     * The operation to convert.
     */
    private final Func1<Observable<T>, Observable<R>> operation;

    /**
     * Constructor.
     * 
     * @param operation
     *            to be converted into {@link Operator}
     */
    public OperationToOperator(Func1<Observable<T>, Observable<R>> operation) {
        this.operation = operation;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super R> subscriber) {
        final PublishSubject<T> subject = PublishSubject.create();
        Subscriber<T> result = createSubscriber(subject);
        subscriber.add(result);
        operation.call(subject).subscribe(subscriber);
        return result;
    }

    /**
     * Creates a subscriber that passes all events on to the subject.
     * 
     * @param subject
     *            receives all events.
     * @return
     */
    private static <T> Subscriber<T> createSubscriber(final PublishSubject<T> subject) {
        return new Subscriber<T>() {

            @Override
            public void onCompleted() {
                subject.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                subject.onError(e);
            }

            @Override
            public void onNext(T t) {
                subject.onNext(t);
            }
        };
    }
}

@@ -130,16 +130,49 @@
 .Func1;%0A
+import rx.observers.Subscribers;%0A
 import r
@@ -459,20 +459,17 @@
 , T%3E %7B%0A%0A
-    
+%09
 public s
@@ -499,32 +499,36 @@
 , T%3E toOperator(
+%0A%09%09%09
 Func1%3CObservable
@@ -556,32 +556,26 @@
 peration) %7B%0A
-        
+%09%09
 return new O
@@ -611,35 +611,26 @@
 ation);%0A
-    %7D%0A%0A    /**%0A    
+%09%7D%0A%0A%09/**%0A%09
  * The o
@@ -650,28 +650,22 @@
 onvert.%0A
-     */%0A    
+%09 */%0A%09
 private 
@@ -718,28 +718,22 @@
 ation;%0A%0A
-    /**%0A    
+%09/**%0A%09
  * Const
@@ -740,28 +740,22 @@
 ructor.%0A
-     * %0A    
+%09 * %0A%09
  * @para
@@ -766,20 +766,17 @@
 eration%0A
-    
+%09
  *      
@@ -819,28 +819,22 @@
 erator%7D%0A
-     */%0A    
+%09 */%0A%09
 public O
@@ -901,24 +901,18 @@
 tion) %7B%0A
-        
+%09%09
 this.ope
@@ -931,27 +931,21 @@
 ration;%0A
-    %7D%0A%0A    
+%09%7D%0A%0A%09
 @Overrid
@@ -946,20 +946,17 @@
 verride%0A
-    
+%09
 public S
@@ -1017,24 +1017,18 @@
 iber) %7B%0A
-        
+%09%09
 final Pu
@@ -1078,24 +1078,18 @@
 eate();%0A
-        
+%09%09
 Subscrib
@@ -1103,22 +1103,16 @@
 esult = 
-create
 Subscrib
@@ -1113,16 +1113,22 @@
 bscriber
+s.from
 (subject
@@ -1130,24 +1130,18 @@
 bject);%0A
-        
+%09%09
 subscrib
@@ -1156,24 +1156,18 @@
 esult);%0A
-        
+%09%09
 operatio
@@ -1209,695 +1209,27 @@
 r);%0A
-        return result;%0A    %7D%0A%0A    /**%0A     * Creates a subscriber that passes all events on to the subject.%0A     * %0A     * @param subject%0A     *            receives all events.%0A     * @return%0A     */%0A    private static %3CT%3E Subscriber%3CT%3E createSubscriber(final PublishSubject%3CT%3E subject) %7B%0A        return new Subscriber%3CT%3E() %7B%0A%0A            @Override%0A            public void onCompleted() %7B%0A                subject.onCompleted();%0A            %7D%0A%0A            @Override%0A            public void onError(Throwable e) %7B%0A                subject.onError(e);%0A            %7D%0A%0A            @Override%0A            public void onNext(T t) %7B%0A                subject.onNext(t);%0A            %7D%0A        %7D;%0A    %7D
+%09%09return result;%0A%09%7D%0A
 %0A%7D%0A

simplify OperatorToOperator