ContentUpdates.java

import android.content.ContentResolver;
import android.database.ContentObserver;
import android.net.Uri;
import android.os.Handler;
import android.os.HandlerThread;
import android.util.Pair;

import java.util.UUID;

import rx.Observable;
import rx.Subscriber;

public class ContentUpdates extends ContentObserver implements Observable.OnSubscribe<Pair<Uri, Boolean>> {
    private static final String TAG = ContentUpdates.class.getSimpleName();
    private Subscriber<? super Pair<Uri, Boolean>> sub;

    private ContentUpdates(final Handler handler) {
        super(handler);
    }

    @Override public void onChange(final boolean selfChange, final Uri uri) {
        if ((sub != null) && !sub.isUnsubscribed()) {
            sub.onNext(new Pair<>(uri, selfChange));
        }
    }

    public static Observable<Pair<Uri, Boolean>> watchUpdates(final ContentResolver resolver, final Uri uri) {
        return Observable.using(
                () -> {
                    final HandlerThread thread = new HandlerThread(TAG + "Looper" + UUID.randomUUID().toString());
                    thread.start();
                    return new Pair<>(thread, new ContentUpdates(new Handler(thread.getLooper())));
                },
                pair -> Observable.unsafeCreate(pair.second)
                                  .doOnSubscribe(() -> resolver.registerContentObserver(uri, true, pair.second)),
                pair -> {
                    resolver.unregisterContentObserver(pair.second);
                    pair.first.quitSafely();
                });
    }

    @Override public void call(final Subscriber<? super Pair<Uri, Boolean>> subscriber) {
        if (sub != null) {
            sub.unsubscribe();
        }
        sub = subscriber;
    }
}