package com.miguelgaeta.bootstrap.mg_preference;

import lombok.AccessLevel;
import lombok.Getter;
import rx.Observable;
import rx.subjects.BehaviorSubject;
import rx.subjects.SerializedSubject;

/**
 * Represents a data object that is automatically exposed as
 * an observable, a blocking getter of the latest or default
 * value and is persisted across subsequent launches via a cache.
 *
 * Created by Miguel Gaeta on 4/2/15.
 */
@SuppressWarnings("UnusedDeclaration")
public class MGPreferenceRx<T> {

    /**
     * Persist value of this data into the
     * android preferences object.
     */
    @Getter(AccessLevel.PRIVATE)
    private MGPreference<T> dataCache;

    /**
     * Can be used by to publish
     * a new value to the data class.
     */
    @Getter(lazy = true, value = AccessLevel.PRIVATE)
    private final SerializedSubject<T, T> dataPublisher = new SerializedSubject<>(BehaviorSubject.create());

    /**
     * Static initializer.
     */
    public static <T> MGPreferenceRx<T> create(String key) {

        return new MGPreferenceRx<>(key, null);
    }

    public static <T> MGPreferenceRx<T> create(String key, T defaultValue) {

        return new MGPreferenceRx<>(key, defaultValue);
    }

    /**
     * Initializes the preference object cache
     * and uses that to initialize rest of the
     * data object.
     */
    private MGPreferenceRx(String key, T defaultValue) {

        if (key != null) {

            // Initialize data cache.
            dataCache = MGPreference.create(key, defaultValue);
        }

        init(defaultValue);
    }

    /**
     * Set the data observable.
     */
    public void set(T t) {

        getDataPublisher().onNext(t);
    }

    /**
     * Get the data publisher as a stream with a generic
     * buffer back-pressure strategy.
     */
    public Observable<T> get(boolean emitNull) {

        Observable<T> stream = getDataPublisher().onBackpressureBuffer();

        if (emitNull) {

            return stream.filter(data -> data != null);
        }

        return stream;
    }

    /**
     * Get the data observable.
     */
    public Observable<T> get() {

        return get(true);
    }

    /**
     * Get blocking version of the data observable.
     */
    public T getBlocking() {

        return get().toBlocking().mostRecent(null).iterator().next();
    }

    /**
     * Initialize the observable data stream
     * and if caching is enabled, set up
     * future value emissions.
     */
    private void init(T defaultValue) {

        get().subscribe(data -> {

            if (getDataCache() != null) {
                getDataCache().set(data);
            }
        });

        // Set initial value from data cache, or just use provided default.
        set(getDataCache() != null ? getDataCache().get() : defaultValue);
    }
}

@@ -114,16 +114,43 @@
 rvable;%0A
+import rx.functions.Func1;%0A
 import r
@@ -1722,32 +1722,671 @@
 Next(t);%0A    %7D%0A%0A
+    public void merge(Func1%3CT, T%3E mergeFunction) %7B%0A%0A        merge(mergeFunction, true);%0A    %7D%0A%0A    public void merge(Func1%3CT, T%3E mergeFunction, boolean emitNull) %7B%0A%0A        merge(mergeFunction, emitNull, true);%0A    %7D%0A%0A    public void merge(Func1%3CT, T%3E mergeFunction, boolean emitNull, boolean mergeOnChange) %7B%0A%0A        get(emitNull).take(1).subscribe(source -%3E %7B%0A%0A            T mergedSource = mergeFunction.call(source);%0A%0A            if (mergeOnChange) %7B%0A%0A                if (source == null ? mergedSource != null : !source.equals(mergedSource)) %7B%0A%0A                    set(mergedSource);%0A                %7D%0A            %7D%0A        %7D);%0A    %7D%0A%0A
     /**%0A     * G

Test implementation of a merge function.