AsyncTask with rxJava

RxAsyncUtils.java

package org.ymkm.utils;

import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;

public final class RxAsyncUtils {

    /**
     * Perform an asynchronous task given input T parameter, yielding an output U type object
     *
     * @param task2 task to execute
     * @param arg the parameter to pass to the task to execute
     * @param <T> input parameter type
     * @param <U> output parameter type
     * @return the resulting rxJava Subscription
     */
    public static <T, U> Subscription run(final AsyncTask2<T, U> task2, final T arg) {
        return Observable.create(new Observable.OnSubscribe<U>() {

            @Override
            public void call(Subscriber<? super U> subscriber) {
                try {
                    final U u = task2.doInBackground(arg);
                    subscriber.onNext(u);
                    subscriber.onCompleted();
                } catch (final Throwable t) {
                    subscriber.onError(t);
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(task2.subscriber());
    }

    /**
     * Perform an asynchronous task, yielding an output T type object.
     *
     * @param task0 the task to execute
     * @param <T> output parameter type
     * @return the resulting rxJava subscription
     */
    public static <T> Subscription run(final AsyncTaskT0<T> task0) {
        return Observable.create(new Observable.OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> subscriber) {
                try {
                    final T t = task0.doInBackground();
                    subscriber.onNext(t);
                    subscriber.onCompleted();
                } catch (final Throwable t) {
                    subscriber.onError(t);
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(task0.subscriber());
    }

    /**
     * Perform an asynchronous task, with no input nor output values
     *
     * @param task0 the task to execute
     * @return the resulting rxJava subscription
     */
    public static Subscription run(final AsyncTask0 task0) {
        return Observable.create(new Observable.OnSubscribe<Byte>() {

            @Override
            public void call(Subscriber<? super Byte> subscriber) {
                try {
                    task0.doInBackground();
                    subscriber.onNext((byte) 0);
                    subscriber.onCompleted();
                } catch (final Throwable t) {
                    subscriber.onError(t);
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(task0.subscriber());
    }

    /**
     * Perform an asynchronous task, given input T type value
     * @param task1 the task to execute
     * @param arg the parameter to pass to the task
     * @param <T> input parameter type
     * @return the resulting rxJava subscription
     */
    public static <T> Subscription run(final AsyncTask1<T> task1, final T arg) {
        return Observable.create(new Observable.OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> subscriber) {
                try {
                    task1.doInBackground(arg);
                    subscriber.onNext(arg);
                    subscriber.onCompleted();
                } catch (final Throwable t) {
                    subscriber.onError(t);
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(task1.subscriber());
    }

    public static abstract class AsyncTask2<T, U> {

        public AsyncTask2() {
        }

        protected abstract U doInBackground(final T arg) throws Throwable;

        protected void onError(final Throwable throwable) {
            throw new RuntimeException(throwable);
        }

        protected void onFinished(final U res) {
        }

        private Subscriber<U> subscriber() {
            return new Subscriber<U>() {

                @Override
                public void onCompleted() {
                }

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

                @Override
                public void onNext(final U res) {
                    AsyncTask2.this.onFinished(res);
                }
            };
        }
    }

    public static abstract class AsyncTask1<T> {

        public AsyncTask1() {
        }

        protected abstract void doInBackground(final T arg) throws Throwable;

        protected void onError(final Throwable throwable) {
            throw new RuntimeException(throwable);
        }

        protected void onFinished(final T arg) {
        }

        private Subscriber<T> subscriber() {
            return new Subscriber<T>() {

                @Override
                public void onCompleted() {
                }

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

                @Override
                public void onNext(final T arg) {
                    AsyncTask1.this.onFinished(arg);
                }
            };
        }

    }

    public static abstract class AsyncTaskT0<T> {

        public AsyncTaskT0() {
        }

        protected abstract T doInBackground() throws Throwable;

        protected void onError(final Throwable throwable) {
            throw new RuntimeException(throwable);
        }

        protected void onFinished(final T t) {
        }


        private Subscriber<T> subscriber() {
            return new Subscriber<T>() {

                @Override
                public void onCompleted() {
                }

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

                @Override
                public void onNext(final T t) {
                    AsyncTaskT0.this.onFinished(t);
                }
            };
        }
    }

    public static abstract class AsyncTask0 {

        public AsyncTask0() {
        }

        protected abstract void doInBackground() throws Throwable;

        protected void onError(final Throwable throwable) {
            throw new RuntimeException(throwable);
        }

        protected void onFinished() {
        }


        private Subscriber<Byte> subscriber() {
            return new Subscriber<Byte>() {

                @Override
                public void onCompleted() {
                }

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

                @Override
                public void onNext(final Byte b) {
                    AsyncTask0.this.onFinished();
                }
            };
        }
    }


    private AsyncUtils() {
    }
}