代码之家 › 专栏 › 技术社区 › Sascha Held

RXJava可观察对象的条件执行+空处理

rx-java2 reactive-programming functional-programming caching android

Sascha Held · 技术社区 · 8 年前

I'm new to the all functional programming and reactive concept and trying to wrapping my head on the following problem.我对整个函数式编程和反应式概念都是新手,我正试图解决以下问题。

我有一个API客户端,我正在对其进行改装。还有一个本地数据库作为API响应的持久缓存。

我想要实现的是:

从本地数据库加载对象
如果没有对象或数据库返回空对象:
- 执行API请求并从联机源获取数据
- 然后,持久化接收到的数据并返回持久化的数据
如果从本地数据库返回了对象,请检查是否需要联机更新
- 需要联机更新(联机获取数据、持久化并返回持久化对象)
- 不需要联机更新(返回本地数据)

我得出的结论如下:

public class LocationCollectionRepository {
private final static Integer fetchInterval = 30; //Minutes
private final LocationService locationService;
private final LocalLocationCollectionRepository localRepository;

public LocationCollectionRepository(@NonNull LocationService locationService, @NonNull LocalLocationCollectionRepository localRepository) {
    this.locationService = locationService;
    this.localRepository = localRepository;
}

public Observable<LocationCollection> getLocationCollection() throws IOException {
    return localRepository.getLocationCollection()
            .takeWhile(this::shouldFetch)
            .flatMap(remoteCollection -> fetchLocationCollection())
            .takeWhile(this::isRequestSuccessful)
            .flatMap(locationCollectionResponse -> persistLocationCollection(locationCollectionResponse.body()));
}

//================================================================================
// Private methods
//================================================================================

private Observable<Response<LocationCollection>> fetchLocationCollection() throws IOException {
    return Observable.fromCallable(() -> {
        LocationServiceQueryBuilder queryBuilder = LocationServiceQueryBuilder.query();
        return queryBuilder.invoke(locationService).execute();
    });
}

private Observable<LocationCollection> persistLocationCollection(@NonNull LocationCollection locationCollection) {
    return localRepository.saveLocationCollection(locationCollection);
}

private boolean shouldFetch(@NonNull Optional<LocationCollection> locationCollection) {
    if (locationCollection.isPresent()) {
        Interval interval = new Interval(new DateTime(locationCollection.get().getTimestamp()), new DateTime());

        return locationCollection.get().getHashValue() == null || interval.toDuration().getStandardMinutes() > fetchInterval;
    } else {
        return true;
    }
}

private boolean isRequestSuccessful(Response<LocationCollection> locationCollectionResponse) throws Exception {
    if (locationCollectionResponse == null || !locationCollectionResponse.isSuccessful()) {
        throw new Exception(locationCollectionResponse.message());
    }
    return true;
}

}

我遇到的问题是,如果数据库返回null,则在订阅回调中不会返回任何对象。我试过了 defaultIfEmpty -方法,但得出的结论是,这也无济于事,因为它期望的是一个对象,而不是一个可观察的对象。

有什么办法,怎么解决这个问题?

2 回复 | 直到 8 年前

toxicafunk 8 年前

您可能应该使用 Flowables 相反无论如何 RxJava 2.x no longer accepts null values and will yield NullPointerException immediately or as a signal to downstream . 如果您切换到Flowables,那么您可以使用 .OneErrorReturnItem(Collections.emptyList()) 这比null提供的信息量要好。没有结果,而不是空值,空值可能意味着不同数量的事情。

Sascha Held 8 年前

我对我的原始答案进行了一些重新评估,得出的结论是,根本不需要对网络响应/从网络获取数据的需要进行大多数就地检查。

首先,如果网络请求出错,将抛出一个异常,该异常将上升到链的上游,并由observable的onError订阅者处理。

其次,也不需要检查请求是否成功,因为通过使用异常,它只能在调用链中的下一步时成功。

第三,使用 takeWhile 使事情变得更加复杂,因为它实际上是需要的。我决定使用一个简单的flatMap Lambda来解决这个问题,该Lambda在内部使用一个非常直接的if语句。因此,我认为代码更具可读性和可理解性。

下面您可以找到我的问题的最终解决方案:

package com.appenetic.fame.model.repository.remote;

import android.support.annotation.NonNull;

import com.annimon.stream.Optional;
import com.appenetic.fame.api.service.LocationService;
import com.appenetic.fame.api.service.LocationServiceQueryBuilder;
import com.appenetic.fame.model.LocationCollection;
import com.appenetic.fame.model.repository.local.LocalLocationCollectionRepository;

import org.joda.time.DateTime;
import org.joda.time.Interval;

import java.io.IOException;

import io.reactivex.Observable;

/**
 * Created by shel on 18.01.18.
 */
public class LocationCollectionRepository {
    private final static Integer fetchInterval = 30; //Minutes
    private final LocationService locationService;
    private final LocalLocationCollectionRepository localRepository;

    public LocationCollectionRepository(@NonNull LocationService locationService, @NonNull LocalLocationCollectionRepository localRepository) {
        this.locationService = locationService;
        this.localRepository = localRepository;
    }

    public Observable<LocationCollection> getLocationCollection() throws IOException {
        return localRepository.getLocationCollection()
                .flatMap(locationCollectionOptional -> {
                    if (shouldFetch(locationCollectionOptional)) {
                        return persistLocationCollection(fetchLocationCollection().blockingFirst());
                    }

                    return Observable.just(locationCollectionOptional.get());
                });
    }

    //================================================================================
    // Private methods
    //================================================================================

    private Observable<LocationCollection> fetchLocationCollection() throws IOException {
        return Observable.fromCallable(() -> {
            LocationServiceQueryBuilder queryBuilder = LocationServiceQueryBuilder.query();
            return queryBuilder.invoke(locationService).execute().body();
        });
    }

    private Observable<LocationCollection> persistLocationCollection(@NonNull LocationCollection locationCollection) {
        return localRepository.saveLocationCollection(locationCollection);
    }

    private boolean shouldFetch(@NonNull Optional<LocationCollection> locationCollection) {
        if (locationCollection.isPresent()) {
            Interval interval = new Interval(new DateTime(locationCollection.get().getTimestamp()), new DateTime());

            return locationCollection.get().getHashValue() == null || interval.toDuration().getStandardMinutes() > fetchInterval;
        } else {
            return true;
        }
    }
}