RxJava Filter implementation for ListView or RecyclerView

BaseFilterPerformer.java

/**
 * Created by Nikola D. on 3/9/2016.
 */
public abstract class BaseFilterPerformer<T> implements FilterOnSubscribe.FilterPerformer<T> {

    public FilterOnSubscribe.FilteringResult<T> createFilterResult(T data) {
        FilterOnSubscribe.FilteringResult<T> filteringResult = new FilterOnSubscribe.FilteringResult<>();
        filteringResult.data = data;
        return filteringResult;
    }
}

FilterOnSubscribe.java

import android.support.annotation.Nullable;

import com.raizlabs.android.dbflow.annotation.NotNull;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func0;

public class FilterOnSubscribe<T> implements Observable.OnSubscribe<FilterOnSubscribe.FilteringResult>, Func0<Observable<FilterOnSubscribe.FilteringResult>> {
    private final FilterPerformer mFilter;
    private String mConstraint;
    private T data;

    protected FilterOnSubscribe(String constraint, @NotNull FilterPerformer performer, @NotNull T in) {
        mConstraint = constraint;
        data = in;
        mFilter = performer;
    }

    @Override
    public void call(Subscriber<? super FilteringResult> subscriber) {
        if (!subscriber.isUnsubscribed()) {
            subscriber.onStart();
        }
        FilteringResult filteringResult = null;
        try {
            filteringResult = mFilter.performFiltering(mConstraint, data);
        } catch (Exception e) {
            if (!subscriber.isUnsubscribed()) {
                subscriber.onError(e);
            }
        }

        if (!subscriber.isUnsubscribed()) {
            subscriber.onCompleted();
        }
        if (!subscriber.isUnsubscribed()) {
            subscriber.onNext(filteringResult);
        }
    }

    @Override
    public Observable<FilteringResult> call() {
        return Observable.create(this);
    }


    public interface FilterPerformer<T> {
        @Nullable
        FilteringResult performFiltering(String constraint, T data);
    }

    public static class FilteringResult<T> {
        public T data;
        public int count;
    }
}

ObservableFilter.java

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

/**
 * Created by Nikola on 3/9/2016.
 */
public class ObservableFilter {
    private static ObservableFilter sInstance;

    public static <T> Observable filter(String constraint, FilterOnSubscribe.FilterPerformer<T> performer, T data) {
        return Observable.defer(newFilter(constraint, performer, data))
                .cache()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    private static <T> FilterOnSubscribe newFilter(String constraint, FilterOnSubscribe.FilterPerformer<T> performer, T data) {
        return new FilterOnSubscribe<>(constraint, performer, data);
    }

    public static ObservableFilter instance() {
        return sInstance == null ? (sInstance = new ObservableFilter()) : sInstance;
    }


    public interface RxFilterable {
        ObservableFilter getRxFilterObservable();
    }

}