Q:

RxAndroid. Simple caching and filtering data

I read a ton of literature about the Rx and, on the one hand, everything is clear, but on the other hand nothing is clear. I'm trying to do a simple filtration and storage of data from the server with this library, and it is not working as expected. I need to implement a simple channel list management:
1. Cache on disk and in memory
2. When user requested - return filtered channels
3. All Subscribers that was attached to this Observable must be notified if filtered cahnnels list was changed (call onNext() for all Subscribers)

I wrote the following:
ArrayList<Channel> channels = null;
ArrayList<Channel> filteredChannels = null;

Observable<ArrayList<Channel>> filteredObservable = Observable.create()
// here I need to check if cache is valid, if no - download from server
// then filter channels and return filteredChannels in onNext call
// all subscribers that called subscribe before and not called unsubscribe must be notified if filteredChannels changed
// how to implement this?
.subscribeOn(Schedulers.io());

public void setFilter(Filter filter) {
// update filteredChannels with the new filter and notify all Subscribers (call onNext()). How to implement this?
}

public Observable<ArrayList<Channel>> getFilteredChannels() {
    return filteredObservable;
}

Do I correctly understood logic of the Rx pattern or not? Thanks in advance.

A:

I found the solution for this task: I need to use flatMap and map and I tried to use BehaviourSubject to notify all subscribers for variable change, but this solution is not stable at all, because of error MissingBackpressureException, that's why I wrote next ObsevableValue class:
public class ObservableValue<T> {
    private static final String TAG = ObservableValue.class.getSimpleName();

    private ArrayList<Registration> subscriptions = new ArrayList<>();
    private T value;
    private int index;
    private boolean firstNotify;
    private ReentrantLock lock = new ReentrantLock();

    public ObservableValue() {
        firstNotify = false;
    }

    public ObservableValue(T defaultValue) {
        value = defaultValue;
        firstNotify = true;
    }

    @SuppressWarnings("unchecked")
    public void setValue(T value) {
        lock.lock();
        this.value = value;
        for (Registration listener: subscriptions) {
            if (!listener.isFinished()) {
                listener.getListener().valueChanged(value);
            }
        }
        lock.unlock();
    }

    public T getValue() {
        lock.lock();
        T result = value;
        lock.unlock();
        return result;
    }

    public Registration subscribe(ValueChangeListener<T> listener) {
        lock.lock();
        Registration s = new Registration(this, index, listener);
        index++;
        subscriptions.add(s);
        if (firstNotify||value!=null) {
            listener.valueChanged(value);
        }
        lock.unlock();
        return s;
    }

    protected void finish(int index) {
        lock.lock();
        for (int i=0;i<subscriptions.size();i++) {
            Registration s = subscriptions.get(i);
            if (s.getIndex()==index) {
                subscriptions.remove(i);
                break;
            }
        }
        lock.unlock();
    }
}

public abstract class ValueChangeListener<T> {
    private Looper looper;

    public ValueChangeListener() {}

    public ValueChangeListener(Looper looper) {
        this.looper = looper;
    }

    public void valueChanged(T data) {
        if (looper!=null) {
            Handler handler = new Handler(looper, msg -> {
                onValueChanged(data);
                return true;
            });
            handler.sendEmptyMessage(0);
        } else {
            onValueChanged(data);
        }
    }

    public abstract void onValueChanged(T data);
}

public class Registration {
    private ObservableValue observableValue;
    private int index;
    private ValueChangeListener listener;
    private volatile boolean finished = false;

    protected Registration(ObservableValue observableValue, int index, ValueChangeListener listener) {
        this.observableValue = observableValue;
        this.index = index;
        this.listener = listener;
    }

    protected ValueChangeListener getListener() {
        return listener;
    }

    protected int getIndex() {
        return index;
    }

    public boolean isFinished() {
        return finished;
    }

    public void finish() {
        finished = true;
        observableValue.finish(index);
    }
}

Now it is working as expected. Like in Rx ObservableValue returns a Registration object that need to be finished() when it is not needed anymore.

