在回调 rxjava 中返回 Observable

Question

我正在搞乱一些 google 感知 api，现在我对 RxJava 的理解限制了我。

我最终想要达到的目标： 我想从 Api 获取天气和位置，并将它们合并到一个对象中，我可以将其传递给我的视图进行更新。

但是，我不确定如何从此处的 api 回调中实现 Observable 的返回，因为它具有 void 返回类型，以及如何实现从 api.getWeather 和 api.getLocation 合并天气和位置对象

public void requestUserCurrentInfo() {
    Subscription userInfo = getWeatherLocation().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).subscribe(userinfo ->
                    Log.d(TAG,userinfo.something()));
}

public Observable<UserInfo> getWeatherLocation () {
    try {
        Awareness.SnapshotApi.getWeather(client)
                .setResultCallback(weather -> {
                    if (!weather.getStatus().isSuccess()) {
                        Log.d(TAG, "Could not get weather");
                        return;
                    }
                    //How do I do here?
                    return weather.getWeather();
                });


        Awareness.SnapshotApi.getLocation(mGoogleApiClient)
            .setResultCallback(retrievedLocation -> {
                if(!retrievedLocation.getStatus().isSuccess()) return;
                Log.d("FRAG", retrievedLocation.getLocation().getLatitude() + "");
            });


    } catch (SecurityException exception) {
        throw new SecurityException("No permission " + exception);

    }

}

对于我项目中的其他事情，我通过遵循存储库模式的 REST api 得到一些东西，然后我可以像这样得到它，因为每一步都返回一个 Observable

getWeatherSubscription = getWeatherUsecase.execute().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).subscribe(
                    smhiResponseModel -> {Log.d(TAG,"Retrieved weather"); locationView.hideLoading();},
                    err -> {Log.d(TAG,"Error fetching weather"); locationView.hideLoading();}
            );

Answer 1

您不会从回调中返回 observable，而是将回调包装到 observable 中以使它们可组合（未经测试）：

    Observable<WeatherResult> weatherObservable = Observable.create(subscriber -> {
        Awareness.SnapshotApi.getWeather(client)
                .setResultCallback(weather -> {
                    if (!weather.getStatus().isSuccess()) {
                        subscriber.onError(new Exception("Could not get weather."));
                        Log.d(TAG, "Could not get weather");
                    } else {
                        //How do I do here?

                        subscriber.onNext(weather);
                        subscriber.onCompleted();
                    }
                });
    });

    Observable<LocationResult> locationObservable = Observable.create(subscriber -> {
        Awareness.SnapshotApi.getLocation(mGoogleApiClient)
                .setResultCallback(retrievedLocation -> {
                    if(!retrievedLocation.getStatus().isSuccess()) {
                        subscriber.onError(new Exception("Could not get location."));
                    } else {
                        Log.d("FRAG", retrievedLocation.getLocation().getLatitude() + "");
                        subscriber.onNext(retrievedLocation);
                        subscriber.onCompleted();
                    }
                });
    });

现在通过.combineLatest() 或.zip() 组合它们：

    Observable<CombinedResult> combinedResults = Observable.zip(weatherObservable, locationObservable,
            (weather, location) -> {
                /* somehow combine weather and location then return as type "CombinedResult" */
            });

别忘了订阅，否则它们都不会被执行：

    combinedResults.subscribe(combinedResult -> {/*do something with that stuff...*/});

Answer 2

Observable.combineLatest(getWeather (), getLocation(), new Func2<List<Object_A>, List<Object_B>, Object>() {
            @Override
            public Object call(Object o, Object o2) {
                combine both results and return the combine result to observer
            }
        })

getweather() 和 getlocation() 返回 observables