package com.github.davidmoten.rx;

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

public class OperatorFromOperation<R, T> implements Operator<R, T> {

	private final Func1<Observable<T>, Observable<R>> operation;

	public OperatorFromOperation(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();
		operation.call(subject).subscribe(subscriber);
		Subscriber<T> result = 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);
			}
		};
		subscriber.add(result);
		return result;
	}
}

@@ -166,16 +166,208 @@
 bject;%0A%0A
+/**%0A * Converts an Operation (a function converting one Observable into another)%0A * into an %7B@link Operator%7D.%0A * %0A * @param %3CR%3E%0A *            to type%0A * @param %3CT%3E%0A *            from type%0A */%0A
 public c
@@ -428,16 +428,56 @@
 , T%3E %7B%0A%0A
+%09/**%0A%09 * The operation to convert.%0A%09 */%0A
 %09private
@@ -531,16 +531,122 @@
 ation;%0A%0A
+%09/**%0A%09 * Constructor.%0A%09 * %0A%09 * @param operation%0A%09 *            to be converted into %7B@link Operator%7D%0A%09 */%0A
 %09public 
@@ -966,16 +966,338 @@
 result =
+ createSubscriber(subject);%0A%09%09subscriber.add(result);%0A%09%09return result;%0A%09%7D%0A%0A%09/**%0A%09 * Creates a subscriber that passes all events on to the subject.%0A%09 * %0A%09 * @param subject%0A%09 *            receives all events.%0A%09 * @return%0A%09 */%0A%09private static %3CT%3E Subscriber%3CT%3E createSubscriber(%0A%09%09%09final PublishSubject%3CT%3E subject) %7B%0A%09%09return
  new Sub
@@ -1544,56 +1544,13 @@
 %09%09%7D%0A
+
 %09%09%7D;%0A
-%09%09subscriber.add(result);%0A%09%09return result;%0A
 %09%7D%0A%7D

add javadoc