RXJava 条件执行 Observables + null 处理

Question

我是整个函数式编程和反应式概念的新手，并试图解决以下问题。

我有一个 API 客户端，我正在为其使用 Retrofit。 还有一个本地数据库充当 API 响应的持久缓存。

我想要实现的是这样的：

1. 从本地数据库加载对象
2. 如果没有对象或数据库返回空对象：
   • 执行 API 请求并从在线来源获取数据
   • 之后，持久化接收到的数据并返回持久化的数据
3. 如果对象从本地数据库返回，请检查是否需要在线更新
   • 需要在线更新（在线获取数据，持久化并返回持久化对象）
   • 不需要在线更新（返回本地数据）

我想出的是以下内容：

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-Method 但得出的结论是这也无济于事，因为它需要一个对象而不是可观察的。

任何想法，如何解决这个问题？

Answer 1

您应该改用Flowables。无论如何， RxJava 2.x no longer accepts null values and will yield NullPointerException immediately or as a signal to downstream。如果您确实切换到 Flowables，那么您可以使用类似 .onErrorReturnItem(Collections.emptyList()) 的东西，它比 null 更好地为您提供多少信息。没有结果，而不是可能意味着不同数量的事物的 null。

Answer 2

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

首先，如果网络请求出现问题，则会抛出异常，该异常将向上链并由 observable 的 onError-subscriber 处理。

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

第三，takeWhile 的使用使事情变得更加复杂，因为它实际上是需要的。 我决定通过使用一个简单的 flatMap-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;
        }
    }
}