/**

* Copyright 2014 Netflix, Inc.

*

* Licensed under the Apache License, Version 2.0 (the "License"); you may not

* use this file except in compliance with the License. You may obtain a copy of

* the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

* License for the specific language governing permissions and limitations under

* the License.

*/

package rx.util.async;

import rx.*;

import rx.Scheduler.Worker;

import rx.functions.*;

import rx.schedulers.Schedulers;

import rx.subjects.AsyncSubject;

import rx.subjects.PublishSubject;

import rx.subjects.Subject;

import rx.subscriptions.SerialSubscription;

import rx.util.async.functions.*;

import rx.util.async.operators.*;

import java.util.concurrent.Callable;

import java.util.concurrent.Future;

import java.util.concurrent.FutureTask;

/**

* Utility methods to convert functions and actions into asynchronous operations through the Observable/Observer

* pattern.

*/

public final class Async {

private Async() {

throw new IllegalStateException("No instances!");

}

/**

* Invokes the specified function asynchronously and returns an Observable that emits the result.

* <p>

* Note: The function is called immediately and once, not whenever an observer subscribes to the resulting

* Observable. Multiple subscriptions to this Observable observe the same return value.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/start.png" alt="">

*

* @param <T> the result value type

* @param func function to run asynchronously

* @return an Observable that emits the function's result value, or notifies observers of an exception

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-start">RxJava Wiki: start()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229036.aspx">MSDN: Observable.Start</a>

*/

public static <T> Observable<T> start(Func0<T> func) {

return startCallable(func);

}

/**

* Invokes the specified function asynchronously and returns an Observable that emits the result.

* <p>

* Note: The function is called immediately and once, not whenever an observer subscribes to the resulting

* Observable. Multiple subscriptions to this Observable observe the same return value.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/start.png" alt="">

*

* @param <R> the result value type

* @param func function to run asynchronously

* @return an Observable that emits the function's result value, or notifies observers of an exception

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-start">RxJava Wiki: start()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229036.aspx">MSDN: Observable.Start</a>

*/

public static <R> Observable<R> startCallable(Callable<? extends R> func) {

return startCallable(func, Schedulers.computation());

}

/**

* Invokes the specified function asynchronously on the specified Scheduler and returns an Observable that

* emits the result.

* <p>

* Note: The function is called immediately and once, not whenever an observer subscribes to the resulting

* Observable. Multiple subscriptions to this Observable observe the same return value.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/start.s.png" alt="">

* <p>

* @param <T> the result value type

* @param func function to run asynchronously

* @param scheduler Scheduler to run the function on

* @return an Observable that emits the function's result value, or notifies observers of an exception

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-start">RxJava Wiki: start()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211721.aspx">MSDN: Observable.Start</a>

*/

public static <T> Observable<T> start(Func0<T> func, Scheduler scheduler) {

return startCallable(func, scheduler);

}

/**

* Invokes the specified callable asynchronously and returns an Observable that emits the result.

* <p>

* Note: The function is called immediately and once, not whenever an observer subscribes to the resulting

* Observable. Multiple subscriptions to this Observable observe the same return value.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/start.png" alt="">

*

* @param <R> the result value type

* @param func function to run asynchronously

* @return an Observable that emits the function's result value, or notifies observers of an exception

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-start">RxJava Wiki: start()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229036.aspx">MSDN: Observable.Start</a>

*/

public static <R> Observable<R> startCallable(final Callable<? extends R> func, final Scheduler scheduler) {

final AsyncSubject<R> subject = AsyncSubject.create();

final Worker inner = scheduler.createWorker();

inner.schedule(new Action0() {

@Override

public void call() {

R result;

try {

result = func.call();

} catch (Throwable t) {

subject.onError(t);

return;

} finally {

inner.unsubscribe();

}

subject.onNext(result);

subject.onCompleted();

}

});

return subject;

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

* <p>

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229868.aspx">MSDN: Observable.ToAsync</a>

*/

public static Func0<Observable<Void>> toAsync(Action0 action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <R> the result value type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229182.aspx">MSDN: Observable.ToAsync</a>

*/

public static <R> Func0<Observable<R>> toAsync(Func0<? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous callable call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <R> the result value type

* @param func the callable to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229182.aspx">MSDN: Observable.ToAsync</a>

*/

public static <R> Func0<Observable<R>> toAsyncThrowing(Callable<? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> first parameter type of the action

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229657.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1> Func1<T1, Observable<Void>> toAsync(Action1<? super T1> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> first parameter type of the action

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229755.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, R> Func1<T1, Observable<R>> toAsync(Func1<? super T1, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> first parameter type of the action

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229755.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, R> Func1<T1, Observable<R>> toAsyncThrowing(ThrowingFunc1<? super T1, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211875.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2> Func2<T1, T2, Observable<Void>> toAsync(Action2<? super T1, ? super T2> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229851.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, R> Func2<T1, T2, Observable<R>> toAsync(Func2<? super T1, ? super T2, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229851.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, R> Func2<T1, T2, Observable<R>> toAsyncThrowing(ThrowingFunc2<? super T1, ? super T2, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229336.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3> Func3<T1, T2, T3, Observable<Void>> toAsync(Action3<? super T1, ? super T2, ? super T3> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229450.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, R> Func3<T1, T2, T3, Observable<R>> toAsync(Func3<? super T1, ? super T2, ? super T3, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229450.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, R> Func3<T1, T2, T3, Observable<R>> toAsyncThrowing(ThrowingFunc3<? super T1, ? super T2, ? super T3, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229769.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4> Func4<T1, T2, T3, T4, Observable<Void>> toAsync(Action4<? super T1, ? super T2, ? super T3, ? super T4> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229911.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, R> Func4<T1, T2, T3, T4, Observable<R>> toAsync(Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229911.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, R> Func4<T1, T2, T3, T4, Observable<R>> toAsyncThrowing(ThrowingFunc4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229577.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5> Func5<T1, T2, T3, T4, T5, Observable<Void>> toAsync(Action5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229571.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, R> Func5<T1, T2, T3, T4, T5, Observable<R>> toAsync(Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229571.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, R> Func5<T1, T2, T3, T4, T5, Observable<R>> toAsyncThrowing(ThrowingFunc5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211773.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6> Func6<T1, T2, T3, T4, T5, T6, Observable<Void>> toAsync(Action6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229716.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, R> Func6<T1, T2, T3, T4, T5, T6, Observable<R>> toAsync(Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229716.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, R> Func6<T1, T2, T3, T4, T5, T6, Observable<R>> toAsyncThrowing(ThrowingFunc6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211812.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7> Func7<T1, T2, T3, T4, T5, T6, T7, Observable<Void>> toAsync(Action7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229773.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, R> Func7<T1, T2, T3, T4, T5, T6, T7, Observable<R>> toAsync(Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229773.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, R> Func7<T1, T2, T3, T4, T5, T6, T7, Observable<R>> toAsyncThrowing(ThrowingFunc7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh228993.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8> Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<Void>> toAsync(Action8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229910.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<R>> toAsync(Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229910.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<R>> toAsyncThrowing(ThrowingFunc8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param <T9> the ninth parameter type

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211702.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8, T9> Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<Void>> toAsync(Action9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9> action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param <T9> the ninth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh212074.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<R>> toAsync(Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <T4> the fourth parameter type

* @param <T5> the fifth parameter type

* @param <T6> the sixth parameter type

* @param <T7> the seventh parameter type

* @param <T8> the eighth parameter type

* @param <T9> the ninth parameter type

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh212074.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<R>> toAsyncThrowing(ThrowingFunc9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.an.png" alt="">

*

* @param action the action to convert

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

*/

public static FuncN<Observable<Void>> toAsync(ActionN action) {

return toAsync(action, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

*/

public static <R> FuncN<Observable<R>> toAsync(FuncN<? extends R> func) {

return toAsync(func, Schedulers.computation());

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.png" alt="">

*

* @param <R> the result type

* @param func the function to convert

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

*/

public static <R> FuncN<Observable<R>> toAsyncThrowing(ThrowingFuncN<? extends R> func) {

return toAsyncThrowing(func, Schedulers.computation());

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.ans.png" alt="">

*

* @param action the action to convert

* @param scheduler the Scheduler used to execute the {@code action}

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229439.aspx">MSDN: Observable.ToAsync</a>

*/

public static Func0<Observable<Void>> toAsync(final Action0 action, final Scheduler scheduler) {

return toAsync(Actions.toFunc(action), scheduler);

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211792.aspx">MSDN: Observable.ToAsync</a>

*/

public static <R> Func0<Observable<R>> toAsync(final Func0<? extends R> func, final Scheduler scheduler) {

return new Func0<Observable<R>>() {

@Override

public Observable<R> call() {

return startCallable(func, scheduler);

}

};

}

/**

* Convert a synchronous callable call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <R> the result type

* @param func the callable to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211792.aspx">MSDN: Observable.ToAsync</a>

*/

public static <R> Func0<Observable<R>> toAsyncThrowing(final Callable<? extends R> func, final Scheduler scheduler) {

return new Func0<Observable<R>>() {

@Override

public Observable<R> call() {

return startCallable(func, scheduler);

}

};

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.ans.png" alt="">

*

* @param <T1> the first parameter type

* @param action the Action to convert

* @param scheduler the Scheduler used to execute the {@code action}

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229822.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1> Func1<T1, Observable<Void>> toAsync(final Action1<? super T1> action, final Scheduler scheduler) {

return toAsync(Actions.toFunc(action), scheduler);

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <T1> the first parameter type

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229731.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, R> Func1<T1, Observable<R>> toAsync(final Func1<? super T1, ? extends R> func, final Scheduler scheduler) {

return new Func1<T1, Observable<R>>() {

@Override

public Observable<R> call(final T1 t1) {

final AsyncSubject<R> subject = AsyncSubject.create();

final Worker inner = scheduler.createWorker();

inner.schedule(new Action0() {

@Override

public void call() {

R result;

try {

result = func.call(t1);

} catch (Throwable t) {

subject.onError(t);

return;

} finally {

inner.unsubscribe();

}

subject.onNext(result);

subject.onCompleted();

}

});

return subject;

}

};

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <T1> the first parameter type

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229731.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, R> Func1<T1, Observable<R>> toAsyncThrowing(final ThrowingFunc1<? super T1, ? extends R> func, final Scheduler scheduler) {

return new Func1<T1, Observable<R>>() {

@Override

public Observable<R> call(T1 t1) {

return startCallable(ThrowingFunctions.toCallable(func, t1), scheduler);

}

};

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.ans.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param action the action to convert

* @param scheduler the Scheduler used to execute the {@code action}

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229722.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2> Func2<T1, T2, Observable<Void>> toAsync(final Action2<? super T1, ? super T2> action, final Scheduler scheduler) {

return toAsync(Actions.toFunc(action), scheduler);

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229327.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, R> Func2<T1, T2, Observable<R>> toAsync(final Func2<? super T1, ? super T2, ? extends R> func, final Scheduler scheduler) {

return new Func2<T1, T2, Observable<R>>() {

@Override

public Observable<R> call(final T1 t1, final T2 t2) {

final AsyncSubject<R> subject = AsyncSubject.create();

final Worker inner = scheduler.createWorker();

inner.schedule(new Action0() {

@Override

public void call() {

R result;

try {

result = func.call(t1, t2);

} catch (Throwable t) {

subject.onError(t);

return;

} finally {

inner.unsubscribe();

}

subject.onNext(result);

subject.onCompleted();

}

});

return subject;

}

};

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229731.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, R> Func2<T1, T2, Observable<R>> toAsyncThrowing(final ThrowingFunc2<? super T1, ? super T2, ? extends R> func, final Scheduler scheduler) {

return new Func2<T1, T2, Observable<R>>() {

@Override

public Observable<R> call(T1 t1, T2 t2) {

return startCallable(ThrowingFunctions.toCallable(func, t1, t2), scheduler);

}

};

}

/**

* Convert a synchronous action call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.ans.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param action the action to convert

* @param scheduler the Scheduler used to execute the {@code action}

* @return a function that returns an Observable that executes the {@code action} and emits {@code null}

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh211787.aspx">MSDN: Observable.ToAsync</a>

*/

public static <T1, T2, T3> Func3<T1, T2, T3, Observable<Void>> toAsync(final Action3<? super T1, ? super T2, ? super T3> action, final Scheduler scheduler) {

return toAsync(Actions.toFunc(action), scheduler);

}

/**

* Convert a synchronous function call into an asynchronous function call through an Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/toAsync.s.png" alt="">

*

* @param <T1> the first parameter type

* @param <T2> the second parameter type

* @param <T3> the third parameter type

* @param <R> the result type

* @param func the function to convert

* @param scheduler the Scheduler used to call the {@code func}

* @return a function that returns an Observable that executes the {@code func} and emits its returned value

* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229287.aspx">MSDN: Observable.ToAsync</a>

*/