public interface SingleOnSubscribe<T> extends Consumer<SingleSubscriber<? super T>> {

}

public interface SingleOperator<Downstream, Upstream> extends Function<SingleSubscriber<? super Downstream>, SingleSubscriber<? super Upstream>> {

}

public interface SingleSubscriber<T> {

void onSubscribe(Disposable d);

void onSuccess(T value);

void onError(Throwable e);

}

public interface SingleTransformer<Upstream, Downstream> extends Function<Single<Upstream>, Single<Downstream>> {

}

public static <T> Single<T> amb(Iterable<? extends Single<? extends T>> sources) {

return create(s -> {

AtomicBoolean once = new AtomicBoolean();

CompositeDisposable set = new CompositeDisposable();

s.onSubscribe(set);

int c = 0;

for (Single<? extends T> s1 : sources) {

if (once.get()) {

return;

}

s1.subscribe(new SingleSubscriber<T>() {

@Override

public void onSubscribe(Disposable d) {

set.add(d);

}

@Override

public void onSuccess(T value) {

if (once.compareAndSet(false, true)) {

s.onSuccess(value);

}

}

@Override

public void onError(Throwable e) {

if (once.compareAndSet(false, true)) {

s.onError(e);

} else {

RxJavaPlugins.onError(e);

}

}

});

c++;

}

if (c == 0 && !set.isDisposed()) {

s.onError(new NoSuchElementException());

}

});

}

@SafeVarargs

@SuppressWarnings("unchecked")

public static <T> Single<T> amb(Single<? extends T>... sources) {

if (sources.length == 0) {

return error(() -> new NoSuchElementException());

}

if (sources.length == 1) {

return (Single<T>)sources[0];

}

return create(s -> {

AtomicBoolean once = new AtomicBoolean();

CompositeDisposable set = new CompositeDisposable();

s.onSubscribe(set);

for (Single<? extends T> s1 : sources) {

if (once.get()) {

return;

}

s1.subscribe(new SingleSubscriber<T>() {

@Override

public void onSubscribe(Disposable d) {

set.add(d);

}

@Override

public void onSuccess(T value) {

if (once.compareAndSet(false, true)) {

s.onSuccess(value);

}

}

@Override

public void onError(Throwable e) {

if (once.compareAndSet(false, true)) {

s.onError(e);

} else {

RxJavaPlugins.onError(e);

}

}

});

}

});

}

public static <T> Observable<T> concat(Iterable<? extends Single<? extends T>> sources) {

return concat(Observable.fromIterable(sources));

}

public static <T> Observable<T> concat(Observable<? extends Single<? extends T>> sources) {

return sources.concatMap(Single::toFlowable);

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

return concat(Observable.fromArray(s1, s2));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

return concat(Observable.fromArray(s1, s2, s3));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

return concat(Observable.fromArray(s1, s2, s3, s4));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

return concat(Observable.fromArray(s1, s2, s3, s4, s5));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

return concat(Observable.fromArray(s1, s2, s3, s4, s5, s6));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

return concat(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7, Single<? extends T> s8

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

return concat(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7, s8));

}

public static <T> Observable<T> concat(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7, Single<? extends T> s8,

Single<? extends T> s9

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

Objects.requireNonNull(s9);

return concat(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7, s8, s9));

}

public static <T> Single<T> create(SingleOnSubscribe<T> onSubscribe) {

Objects.requireNonNull(onSubscribe);

// TODO plugin wrapper

return new Single<>(onSubscribe);

}

public static <T> Single<T> defer(Supplier<? extends Single<? extends T>> singleSupplier) {

return create(s -> {

Single<? extends T> next;

try {

next = singleSupplier.get();

} catch (Throwable e) {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(e);

return;

}

if (next == null) {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(new NullPointerException("The Single supplied was null"));

return;

}

next.subscribe(s);

});

}

public static <T> Single<T> error(Supplier<? extends Throwable> errorSupplier) {

Objects.requireNonNull(errorSupplier);

return create(s -> {

Throwable error;

try {

error = errorSupplier.get();

} catch (Throwable e) {

error = e;

}

if (error == null) {

error = new NullPointerException();

}

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(error);

});

}

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

Objects.requireNonNull(error);

return error(() -> error);

}

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

Objects.requireNonNull(callable);

return create(s -> {

s.onSubscribe(EmptyDisposable.INSTANCE);

try {

T v = callable.call();

if (v != null) {

s.onSuccess(v);

} else {

s.onError(new NullPointerException());

}

} catch (Throwable e) {

s.onError(e);

}

});

}

public static <T> Single<T> fromFuture(CompletableFuture<? extends T> future) {

Objects.requireNonNull(future);

return create(s -> {

BooleanDisposable bd = new BooleanDisposable(() -> future.cancel(true));

s.onSubscribe(bd);

future.whenComplete((v, e) -> {

if (e != null) {

if (!bd.isDisposed()) {

s.onError(e);

}

} else {

if (v != null) {

s.onSuccess(v);

} else {

s.onError(new NullPointerException());

}

}

});

});

}

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

return Observable.<T>fromFuture(future).toSingle();

}

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

return Observable.<T>fromFuture(future, timeout, unit).toSingle();

}

public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit, Scheduler scheduler) {

return Observable.<T>fromFuture(future, timeout, unit, scheduler).toSingle();

}

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

return Observable.<T>fromFuture(future, scheduler).toSingle();

}

public static <T> Single<T> fromPublisher(Publisher<? extends T> publisher) {

Objects.requireNonNull(publisher);

return create(s -> {

publisher.subscribe(new Subscriber<T>() {

T value;

@Override

public void onComplete() {

T v = value;

value = null;

if (v != null) {

s.onSuccess(v);

} else {

s.onError(new NoSuchElementException());

}

}

@Override

public void onError(Throwable t) {

value = null;

s.onError(t);

}

@Override

public void onNext(T t) {

value = t;

}

@Override

public void onSubscribe(Subscription inner) {

s.onSubscribe(inner::cancel);

inner.request(Long.MAX_VALUE);

}

});

});

}

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

Objects.requireNonNull(value);

return create(s -> {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onSuccess(value);

});

}

public static <T> Observable<T> merge(Iterable<? extends Single<? extends T>> sources) {

return merge(Observable.fromIterable(sources));

}

public static <T> Observable<T> merge(Observable<? extends Single<? extends T>> sources) {

return sources.flatMap(Single::toFlowable);

}

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

return source.flatMap(v -> v);

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

return merge(Observable.fromArray(s1, s2));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

return merge(Observable.fromArray(s1, s2, s3));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

return merge(Observable.fromArray(s1, s2, s3, s4));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

return merge(Observable.fromArray(s1, s2, s3, s4, s5));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

return merge(Observable.fromArray(s1, s2, s3, s4, s5, s6));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

return merge(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7, Single<? extends T> s8

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

return merge(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7, s8));

}

public static <T> Observable<T> merge(

Single<? extends T> s1, Single<? extends T> s2,

Single<? extends T> s3, Single<? extends T> s4,

Single<? extends T> s5, Single<? extends T> s6,

Single<? extends T> s7, Single<? extends T> s8,

Single<? extends T> s9

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

Objects.requireNonNull(s9);

return merge(Observable.fromArray(s1, s2, s3, s4, s5, s6, s7, s8, s9));

}

public static <T> Single<T> never() {

return create(s -> {

s.onSubscribe(EmptyDisposable.INSTANCE);

});

}

public static Single<Long> timer(long delay, TimeUnit unit) {

return timer(delay, unit, Schedulers.computation());

}

public static Single<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {

Objects.requireNonNull(unit);

Objects.requireNonNull(scheduler);

return create(s -> {

MultipleAssignmentDisposable mad = new MultipleAssignmentDisposable();

s.onSubscribe(mad);

mad.set(scheduler.scheduleDirect(() -> s.onSuccess(0L), delay, unit));

});

}

public static <T> Single<Boolean> equals(Single<? extends T> first, Single<? extends T> second) {

Objects.requireNonNull(first);

Objects.requireNonNull(second);

return create(s -> {

AtomicInteger count = new AtomicInteger();

Object[] values = { null, null };

CompositeDisposable set = new CompositeDisposable();

s.onSubscribe(set);

class InnerSubscriber implements SingleSubscriber<T> {

final int index;

public InnerSubscriber(int index) {

this.index = index;

}

@Override

public void onSubscribe(Disposable d) {

set.add(d);

}

@Override

public void onSuccess(T value) {

values[index] = value;

if (count.incrementAndGet() == 2) {

s.onSuccess(Objects.equals(values[0], values[1]));

}

}

@Override

public void onError(Throwable e) {

for (;;) {

int state = count.get();

if (state >= 2) {

RxJavaPlugins.onError(e);

return;

}

if (count.compareAndSet(state, 2)) {

s.onError(e);

return;

}

}

}

}

first.subscribe(new InnerSubscriber(0));

second.subscribe(new InnerSubscriber(1));

});

}

@SuppressWarnings("unchecked")

private static <T1, T2, R> Function<Object[], R> toFunction(BiFunction<? super T1, ? super T2, ? extends R> biFunction) {

Objects.requireNonNull(biFunction);

return a -> {

if (a.length != 2) {

throw new IllegalArgumentException("Array of size 2 expected but got " + a.length);

}

return ((BiFunction<Object, Object, R>)biFunction).apply(a[0], a[1]);

};

}

public static <T, U> Single<T> using(Supplier<U> resourceSupplier,

Function<? super U, ? extends Single<? extends T>> singleFunction, Consumer<? super U> disposer) {

return using(resourceSupplier, singleFunction, disposer);

}

public static <T, U> Single<T> using(Supplier<U> resourceSupplier,

Function<? super U, ? extends Single<? extends T>> singleFunction, Consumer<? super U> disposer, boolean eager) {

Objects.requireNonNull(resourceSupplier);

Objects.requireNonNull(singleFunction);

Objects.requireNonNull(disposer);

return create(s -> {

U resource;

try {

resource = resourceSupplier.get();

} catch (Throwable ex) {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(ex);

return;

}

Single<? extends T> s1;

try {

s1 = singleFunction.apply(resource);

} catch (Throwable ex) {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(ex);

return;

}

if (s1 == null) {

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(new NullPointerException("The Single supplied by the function was null"));

return;

}

s1.subscribe(new SingleSubscriber<T>() {

@Override

public void onSubscribe(Disposable d) {

if (eager) {

CompositeDisposable set = new CompositeDisposable();

set.add(d);

set.add(() -> {

try {

disposer.accept(resource);

} catch (Throwable e) {

RxJavaPlugins.onError(e);

}

});

} else {

s.onSubscribe(d);

}

}

@Override

public void onSuccess(T value) {

if (eager) {

try {

disposer.accept(resource);

} catch (Throwable e) {

s.onError(e);

return;

}

}

s.onSuccess(value);

if (!eager) {

try {

disposer.accept(resource);

} catch (Throwable e) {

RxJavaPlugins.onError(e);

}

}

}

@Override

public void onError(Throwable e) {

if (eager) {

try {

disposer.accept(resource);

} catch (Throwable ex) {

e.addSuppressed(ex);

}

}

s.onError(e);

if (!eager) {

try {

disposer.accept(resource);

} catch (Throwable ex) {

e.addSuppressed(ex);

RxJavaPlugins.onError(e);

}

}

}

});

});

}

public static <T, R> Observable<R> zip(Iterable<? extends Single<? extends T>> sources, Function<? super Object[], ? extends R> zipper) {

Iterable<? extends Observable<T>> it = () -> {

Iterator<? extends Single<? extends T>> sit = sources.iterator();

return new Iterator<Observable<T>>() {

@Override

public boolean hasNext() {

return sit.hasNext();

}

@SuppressWarnings("unchecked")

@Override

public Observable<T> next() {

return ((Observable<T>)sit.next().toFlowable());

}

};

};

return Observable.zipIterable(zipper, false, 1, it);

}

public static <T1, T2, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

BiFunction<? super T1, ? super T2, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

return zipArray(toFunction(zipper), s1, s2);

}

public static <T1, T2, T3, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3,

Function3<? super T1, ? super T2, ? super T3, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

return zipArray(zipper, s1, s2, s3);

}

public static <T1, T2, T3, T4, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Function4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

return zipArray(zipper, s1, s2, s3, s4);

}

public static <T1, T2, T3, T4, T5, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Single<? extends T5> s5,

Function5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

return zipArray(zipper, s1, s2, s3, s4, s5);

}

public static <T1, T2, T3, T4, T5, T6, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Single<? extends T5> s5, Single<? extends T6> s6,

Function6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

return zipArray(zipper, s1, s2, s3, s4, s5, s6);

}

public static <T1, T2, T3, T4, T5, T6, T7, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Single<? extends T5> s5, Single<? extends T6> s6,

Single<? extends T7> s7,

Function7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

return zipArray(zipper, s1, s2, s3, s4, s5, s6, s7);

}

public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Single<? extends T5> s5, Single<? extends T6> s6,

Single<? extends T7> s7, Single<? extends T8> s8,

Function8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

return zipArray(zipper, s1, s2, s3, s4, s5, s6, s7, s8);

}

public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> zip(

Single<? extends T1> s1, Single<? extends T2> s2,

Single<? extends T3> s3, Single<? extends T4> s4,

Single<? extends T5> s5, Single<? extends T6> s6,

Single<? extends T7> s7, Single<? extends T8> s8,

Single<? extends T9> s9,

Function9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> zipper

) {

Objects.requireNonNull(s1);

Objects.requireNonNull(s2);

Objects.requireNonNull(s3);

Objects.requireNonNull(s4);

Objects.requireNonNull(s5);

Objects.requireNonNull(s6);

Objects.requireNonNull(s7);

Objects.requireNonNull(s8);

Objects.requireNonNull(s9);

return zipArray(zipper, s1, s2, s3, s4, s5, s6, s7, s8, s9);

}

@SafeVarargs

@SuppressWarnings({"rawtypes", "unchecked"})

public static <T, R> Observable<R> zipArray(Function<? super Object[], ? extends R> zipper, Single<? extends T>... sources) {

Publisher[] sourcePublishers = new Publisher[sources.length];

int i = 0;

for (Single<? extends T> s : sources) {

sourcePublishers[i] = s.toFlowable();

i++;

}

return Observable.zipArray(zipper, false, 1, sourcePublishers);

}

protected final SingleOnSubscribe<T> onSubscribe;

protected Single(SingleOnSubscribe<T> onSubscribe) {

this.onSubscribe = onSubscribe;

}

public final Single<T> ambWith(Single<? extends T> other) {

return amb(this, other);

}

public final Single<T> asSingle() {

return create(s -> subscribe(s));

}

public final <R> Single<R> compose(Function<? super Single<T>, ? extends Single<R>> convert) {

return to(convert);

}

public final Single<T> cache() {

AtomicInteger wip = new AtomicInteger();

AtomicReference<Object> notification = new AtomicReference<>();

List<SingleSubscriber<? super T>> subscribers = new ArrayList<>();

return create(s -> {

Object o = notification.get();

if (o != null) {

s.onSubscribe(EmptyDisposable.INSTANCE);

if (NotificationLite.isError(o)) {

s.onError(NotificationLite.getError(o));

} else {

s.onSuccess(NotificationLite.getValue(o));

}

return;

}

synchronized (subscribers) {

o = notification.get();

if (o == null) {

subscribers.add(s);

}

}

if (o != null) {

s.onSubscribe(EmptyDisposable.INSTANCE);

if (NotificationLite.isError(o)) {

s.onError(NotificationLite.getError(o));

} else {

s.onSuccess(NotificationLite.getValue(o));

}

return;

}

if (wip.getAndIncrement() != 0) {

return;

}

subscribe(new SingleSubscriber<T>() {

@Override

public void onSubscribe(Disposable d) {

}

@Override

public void onSuccess(T value) {

notification.set(NotificationLite.next(value));

List<SingleSubscriber<? super T>> list;

synchronized (subscribers) {

list = new ArrayList<>(subscribers);

subscribers.clear();

}

for (SingleSubscriber<? super T> s1 : list) {

s1.onSuccess(value);

}

}

@Override

public void onError(Throwable e) {

notification.set(NotificationLite.error(e));

List<SingleSubscriber<? super T>> list;

synchronized (subscribers) {

list = new ArrayList<>(subscribers);

subscribers.clear();

}

for (SingleSubscriber<? super T> s1 : list) {

s1.onError(e);

}

}

});

});

}

public final <U> Single<U> cast(Class<? extends U> clazz) {

return create(s -> {

subscribe(new SingleSubscriber<T>() {

@Override

public void onSubscribe(Disposable d) {

s.onSubscribe(d);

}

@Override

public void onSuccess(T value) {

U v;

try {

v = clazz.cast(value);

} catch (ClassCastException ex) {

s.onError(ex);

return;

}

s.onSuccess(v);

}

@Override

public void onError(Throwable e) {

s.onError(e);

}

});

});

}

public final Observable<T> concatWith(Single<? extends T> other) {

return concat(this, other);

}