java - 更新缓存 rxjava observables 数据(使用嵌套缓存!)

Question

我编写了一个小测试来重现我的问题。我想嵌套缓存的可观察量并反射(reflect)对原始对象所做的底层数据结构所做的所有更改，但使用嵌套缓存这似乎会失败。

问题能解决吗？我已经在问题下写下了我的期望......

测试

public static void test()
{
    // --------------------
    // TEST 1 - simple caching
    // --------------------

    List<Data> emptyData1 = getEmptyData()
            .toBlocking()
            .single();

    L.d("TEST 1a - Size: %d == %d", emptyData1.size(), 0);

    // add 1 empty data
    emptyData1.add(new Data(0, false));
    L.d("TEST 1b - Size: %d == %d (Loaded: %b)", emptyData1.size(), 1, emptyData1.get(0).loaded);

    List<Data> emptyData2 = getEmptyData()
            .toBlocking()
            .single();

    L.d("TEST 1c - Size: %d == %d (Loaded: %b)", emptyData2.size(), 1, emptyData2.get(0).loaded);

    // --------------------
    // TEST 2 - nested caching
    // --------------------

    List<Data> loadedData1 = getLoadedData()
            .toBlocking()
            .single();

    L.d("TEST 2a - Size: %d == %d (Loaded: %b)", loadedData1.size(), 1, loadedData1.get(0).loaded);

    // add a second data, this time we add a loaded data
    loadedData1.add(new Data(1, true));

    List<Data> loadedData2 = getLoadedData()
            .toBlocking()
            .single();

    L.d("TEST 2b - Size: %d == %d (Loaded: %b, %b)", loadedData1.size(), 2, loadedData2.get(0).loaded, loadedData2.get(1).loaded);

    // --------------------
    // TEST 3 - test if empty observable is holding the loaded data as well now
    // --------------------

    List<Data> testEmptyStateData = getEmptyData()
            .toBlocking()
            .single();

    L.d("TEST 3a - Size: %d == %d", testEmptyStateData.size(), 2);

    // I don't expect this, but this will happen
    if (testEmptyStateData.size() == 1)
    {
        L.d("TEST 3b1 - Size: %d == %d (Loaded: %b)", testEmptyStateData.size(), 2, testEmptyStateData.get(0).loaded);
    }
    // I expect this, but it won't be true
    if (testEmptyStateData.size() == 2)
    {
        L.d("TEST 3b - Size: %d == %d (Loaded: %b, %b)", testEmptyStateData.size(), 2, testEmptyStateData.get(0).loaded, testEmptyStateData.get(1).loaded);
    }
}

结果

TEST 1a - Size: 0 == 0
TEST 1b - Size: 1 == 1 (Loaded: false)
TEST 1c - Size: 1 == 1 (Loaded: false)

TEST 2a - Size: 1 == 1 (Loaded: true)
TEST 2b - Size: 2 == 2 (Loaded: true, true)

TEST 3a - Size: 1 == 2 // <= UNDESIRED RESULT!!!
TEST 3b1 - Size: 1 == 2 (Loaded: true) // <= at least the object in the list is correct! But I would expect that the empty observalbe would hold 2 items, both loaded! "Test 3b2" should be printed instead

问题

我希望所有可观察量都会将原始列表传递到流中，并且我总是得到原始对象，并且我对此对象的所有更改都将反射(reflect)在基本列表中，因此最后是空的可观察量应返回 2 个已加载的数据项，但对于此嵌套缓存场景而言并非如此。

我需要什么

• 每个操作只能运行一次!无论是否有人订阅了可观察的内容!这就是为什么我决定使用 cache()
• 可观察量必须可共享且必须可重用

代码

辅助函数和数据

private static  Observable<List<Data>> mEmptyObservable = null;
private static  Observable<List<Data>> mLoadedObservable = null;

public static Observable<List<Data>> getEmptyData()
{
    if (mEmptyObservable == null)
    {
        // simple test data
        List<Data> values = new ArrayList<>();
        mEmptyObservable = Observable.just(values)
                // cache and share observable
                .cache().replay().refCount();
    }
    return mEmptyObservable;
}

public static Observable<List<Data>> getLoadedData()
{
    if (mLoadedObservable == null)
    {
        mLoadedObservable = getEmptyData()
                .flatMap(new Func1<List<Data>, Observable<Data>>()
                {
                    @Override
                    public Observable<Data> call(List<Data> datas)
                    {
                        return Observable.from(datas);
                    }
                })
                .map(new Func1<Data, Data>()
                {
                    @Override
                    public Data call(Data data)
                    {
                        data.load();
                        return data;
                    }
                })
                .toList()
                // cache and share observable
                .cache().replay().refCount();
    }
    return mLoadedObservable;
}

数据类

static class Data
{
    int index;
    boolean loaded;

    public Data(int index, boolean processed)
    {
        this.index = index;
        this.loaded = processed;
    }

    public void load()
    {
        if (!loaded)
        {
            // do some have operation... once only per data!
        }
        loaded = true;
    }
}

Answer 1

您同时使用了 cache() 和 replay()。尝试类似的事情

private static  Observable<List<Data>> mEmptyObservable = null;
private static  Observable<List<Data>> mLoadedObservable = null;

public static Observable<List<Data>> getEmptyData()
{
    if (mEmptyObservable == null)
    {
        // simple test data
        List<Data> values = new ArrayList<>();
        mEmptyObservable = Observable.just(values)
                // share and replay observable
                .share().replay();
    }
    return mEmptyObservable;
}

public static Observable<List<Data>> getLoadedData()
{
    if (mLoadedObservable == null)
    {
        mLoadedObservable = getEmptyData()
                .flatMap(new Func1<List<Data>, Observable<Data>>()
                {
                    @Override
                    public Observable<Data> call(List<Data> datas)
                    {
                        return Observable.from(datas);
                    }
                })
                .map(new Func1<Data, Data>()
                {
                    @Override
                    public Data call(Data data)
                    {
                        data.load();
                        return data;
                    }
                })
                .toList()
                // share and replay observable
                .share().replay();
    }
    return mLoadedObservable;
}

this answer中有更多信息.

编辑:然而，真正的问题似乎是您在 getLoadedData() 中创建了不同的列表。如果您想要相同的列表，请尝试类似

public static Observable<List<Data>> getLoadedData()
{
    if (mLoadedObservable == null)
    {
        mLoadedObservable = getEmptyData().
                .map(new Func1<List<Data>, List<Data>>()
                {
                    @Override
                    public Data call(List<Data> data)
                    {
                        for (Data item : data) {
                           item.load();
                        }
                        return data;
                    }
                })
                // share and replay observable
                .share().replay();
    }
    return mLoadedObservable;
}