PublishSubjectLiveDate.java

/**
 * Call onChange only data set after observe (do not call onChange for old data on call observe method)
 */
public class PublishSubjectLiveDate<T> extends MutableLiveData<T> {

    private static final String TAG = "PublishSubjectLiveDate";

    private AtomicInteger version = new AtomicInteger();
    private HashMap<Observer<? super T>, ObserverWrapper<T>> wrappers = new HashMap<>();

    public AtomicInteger getVersion() {
        return version;
    }

    @MainThread
    public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
        ObserverWrapper<T> observerWrapper = new ObserverWrapper<>(observer, this);
        wrappers.put(observer, observerWrapper);
        super.observe(owner, observerWrapper);
    }

    @MainThread
    public void observeForever(@NonNull Observer<? super T> observer) {
        ObserverWrapper<T> observerWrapper = new ObserverWrapper<>(observer, this);
        wrappers.put(observer, observerWrapper);
        super.observeForever(observerWrapper);
    }

    @Override
    public void removeObserver(@NonNull Observer<? super T> observer) {

        if (observer instanceof ObserverWrapper) {
            super.removeObserver(observer);
            Observer<? super T> key = null;
            for (Map.Entry<Observer<? super T>, ObserverWrapper<T>> entry : wrappers.entrySet()) {
                if (entry.getValue() == observer) {
                    key = entry.getKey();
                    break;
                }
            }
            if (key != null) {
                wrappers.remove(key);
            }

        } else if (wrappers.containsKey(observer)) {
            ObserverWrapper observerWrapper = wrappers.get(observer);
            if (observerWrapper != null) {
                super.removeObserver(observerWrapper);
            }
            wrappers.remove(observer);
        }
    }


    @MainThread
    public void setValue(@Nullable T t) {
        version.incrementAndGet();
        super.setValue(t);
    }

    /**
     * Used for cases where T is Void, to make calls cleaner.
     */
    @MainThread
    public void call() {
        setValue(null);
    }

    public void callFromThread() {
        super.postValue(null);
    }

    private static class ObserverWrapper<T> implements Observer<T> {
        private final Observer<? super T> delegateObserver;
        private final PublishSubjectLiveDate<T> liveDate;
        private int lastVersion;


        public ObserverWrapper(Observer<? super T> delegateObserver, PublishSubjectLiveDate<T> liveDate) {
            this.delegateObserver = delegateObserver;
            this.liveDate = liveDate;
            this.lastVersion = liveDate.getVersion().get();
        }

        @Override
        public void onChanged(T t) {
            int liveDataVersion = liveDate.getVersion().get();
            if (liveDataVersion > lastVersion) {
                lastVersion = liveDataVersion;
                delegateObserver.onChanged(t);
            }
        }
    }

}