A util class for RxJava which provides Observable sources for Cursor, LocalBroadcastManager and SparseArray

RxUtils.java

import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.database.Cursor;
import android.support.v4.content.LocalBroadcastManager;
import android.support.v4.util.LongSparseArray;
import android.util.SparseArray;

import io.reactivex.Observable;
import io.reactivex.functions.Function;

public final class RxUtils {

    public static <T> Observable<T> fromCursor(Cursor cursor, Function<Cursor, T> parser) {
        return Observable.create($ -> {
            try {
                while (!$.isDisposed() && cursor.moveToNext()) {
                    T value = parser.apply(cursor);

                    if (value != null) {
                        $.onNext(value);
                    }
                }
                $.onComplete();

            } finally {
                cursor.close();
            }
        });
    }

    public static Observable<Intent> fromLocalBroadcast(Context context, String action, String... others) {
        IntentFilter filter = new IntentFilter(action);
        for (String other : others) {
            filter.addAction(other);
        }
        return fromLocalBroadcast(context, filter);
    }

    public static Observable<Intent> fromLocalBroadcast(Context context, IntentFilter filter) {
        LocalBroadcastManager manager = LocalBroadcastManager.getInstance(context);

        return Observable.create($ -> {
            BroadcastReceiver receiver = new BroadcastReceiver() {

                @Override
                public void onReceive(Context context, Intent intent) {
                    $.onNext(intent);
                }

            };

            $.setCancellable(() -> manager.unregisterReceiver(receiver));

            manager.registerReceiver(receiver, filter);
        });
    }

    public static <T> Observable<T> fromSparse(SparseArray<T> array) {
        return Observable.create($ -> {
            for (int i = 0, c = array.size(); i < c; i++) {
                $.onNext(array.valueAt(i));
            }
            $.onComplete();
        });
    }

    public static <T> Observable<T> fromSparse(LongSparseArray<T> array) {
        return Observable.create($ -> {
            for (int i = 0, c = array.size(); i < c; i++) {
                $.onNext(array.valueAt(i));
            }
            $.onComplete();
        });
    }

    private RxUtils() {
    }

}