final Observable.OnSubscribe<T> onSubscribe;

/**

protected Single(final OnSubscribe<T> f) {

// bridge between OnSubscribe (which all Operators and Observables use) and OnExecute (for Single)

this.onSubscribe = new Observable.OnSubscribe<T>() {

@Override

public void call(final Subscriber<? super T> child) {

final SingleDelayedProducer<T> producer = new SingleDelayedProducer<T>(child);

child.setProducer(producer);

SingleSubscriber<T> ss = new SingleSubscriber<T>() {

@Override

public void onSuccess(T value) {

producer.setValue(value);

}

@Override

public void onError(Throwable error) {

child.onError(error);

}

};

child.add(ss);

f.call(ss);

}

};

}

private Single(final Observable.OnSubscribe<T> f) {

this.onSubscribe = f;

}

private static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();

/**

public final static <T> Single<T> create(OnSubscribe<T> f) {

return new Single<T>(f); // TODO need hook

}

/**

public interface OnSubscribe<T> extends Action1<SingleSubscriber<? super T>> {

// cover for generics insanity

}

/**

private final <R> Single<R> lift(final Operator<? extends R, ? super T> lift) {

// This method is private because not sure if we want to expose the Observable.Operator in this public API rather than a Single.Operator

return new Single<R>(new Observable.OnSubscribe<R>() {

@Override

public void call(Subscriber<? super R> o) {

try {

final Subscriber<? super T> st = hook.onLift(lift).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 (OnErrorNotImplementedException) e;

}

st.onError(e);

}

} catch (Throwable e) {

if (e instanceof OnErrorNotImplementedException) {

throw (OnErrorNotImplementedException) 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);

}

}

});

}

/**

@SuppressWarnings("unchecked")

public <R> Single<R> compose(Transformer<? super T, ? extends R> transformer) {

return ((Transformer<T, R>) transformer).call(this);

}

/**

public interface Transformer<T, R> extends Func1<Single<T>, Single<R>> {

// cover for generics insanity

}

/**

private static <T> Observable<T> asObservable(Single<T> t) {

// is this sufficient, or do I need to keep the outer Single and subscribe to it?

return Observable.create(t.onSubscribe);

}

/**

private final Single<Observable<T>> nest() {

return Single.just(asObservable(this));

}

/* *********************************************************************************************************

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2) {

return Observable.concat(asObservable(t1), asObservable(t2));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7, Single<? extends T> t8) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7), asObservable(t8));

}

/**

public final static <T> Observable<T> concat(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7, Single<? extends T> t8, Single<? extends T> t9) {

return Observable.concat(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7), asObservable(t8), asObservable(t9));

}

/**

public final static <T> Single<T> error(final Throwable exception) {

return Single.create(new OnSubscribe<T>() {

@Override

public void call(SingleSubscriber<? super T> te) {

te.onError(exception);

}

});

}

/**

public final static <T> Single<T> from(Future<? extends T> future) {

return new Single<T>(OnSubscribeToObservableFuture.toObservableFuture(future));

}

/**

public final static <T> Single<T> from(Future<? extends T> future, long timeout, TimeUnit unit) {

return new Single<T>(OnSubscribeToObservableFuture.toObservableFuture(future, timeout, unit));

}

/**

public final static <T> Single<T> from(Future<? extends T> future, Scheduler scheduler) {

return new Single<T>(OnSubscribeToObservableFuture.toObservableFuture(future)).subscribeOn(scheduler);

}

/**

@Experimental

public static <T> Single<T> fromCallable(final Callable<? extends T> func) {

return create(new OnSubscribe<T>() {

@Override

public void call(SingleSubscriber<? super T> singleSubscriber) {

final T value;

try {

value = func.call();

} catch (Throwable t) {

Exceptions.throwIfFatal(t);

singleSubscriber.onError(t);

return;

}

singleSubscriber.onSuccess(value);

}

});

}

/**

public final static <T> Single<T> just(final T value) {

// TODO add similar optimization as ScalarSynchronousObservable

return Single.create(new OnSubscribe<T>() {

@Override

public void call(SingleSubscriber<? super T> te) {

te.onSuccess(value);

}

});

}

/**

public final static <T> Single<T> merge(final Single<? extends Single<? extends T>> source) {

return Single.create(new OnSubscribe<T>() {

@Override

public void call(final SingleSubscriber<? super T> child) {

source.subscribe(new SingleSubscriber<Single<? extends T>>() {

@Override

public void onSuccess(Single<? extends T> innerSingle) {

innerSingle.subscribe(child);

}

@Override

public void onError(Throwable error) {

child.onError(error);

}

});

}

});

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2) {

return Observable.merge(asObservable(t1), asObservable(t2));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7, Single<? extends T> t8) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7), asObservable(t8));

}

/**

public final static <T> Observable<T> merge(Single<? extends T> t1, Single<? extends T> t2, Single<? extends T> t3, Single<? extends T> t4, Single<? extends T> t5, Single<? extends T> t6, Single<? extends T> t7, Single<? extends T> t8, Single<? extends T> t9) {

return Observable.merge(asObservable(t1), asObservable(t2), asObservable(t3), asObservable(t4), asObservable(t5), asObservable(t6), asObservable(t7), asObservable(t8), asObservable(t9));

}

/**

public final static <T1, T2, R> Single<R> zip(Single<? extends T1> o1, Single<? extends T2> o2, final Func2<? super T1, ? super T2, ? extends R> zipFunction) {

return just(new Observable<?>[] { asObservable(o1), asObservable(o2) }).lift(new OperatorZip<R>(zipFunction));

}

/**