SearchFragment.java

    @Override
    public void onAttach(final Activity activity) {
        super.onAttach(activity);
        mContextStream.onNext(activity);
    }


// ......


        rx.Observable.combineLatest(mResultObservable,
                // Activity
                mContextStream.filter(new Func1<Activity, Boolean>() {
                    @Override
                    public Boolean call(Activity activity) {
                        return activity != null;
                    }
                }),

                // Combine
                new Func2<ContentValues[], Activity, Pair<ContentValues[], Activity>>() {
                    @Override
                    public Pair<ContentValues[], Activity> call(ContentValues[] values, Activity a) {
                        return new Pair<>(values, a);
                    }
                }

        ).subscribe(new Action1<Pair<ContentValues[], Activity>>() {
            @Override
            public void call(Pair<ContentValues[], Activity> pair) {
                pair.second.getContentResolver().bulkInsert(
                        DbContract.SearchEntry.SEARCH_URI,
                        pair.first);
            }
        });