标签: rx-java

我需要在 TextView 中显示百分比值，例如“15 %”，我通过这种方式在 viewModel 中执行此操作，并且它有效：

                    perfusionIndexValue.postValue(
                        list.lastOrNull()?.perfusionIndex?.takeIf {
                            it != PerfusionIndex.NO_DATA
                        }?.value?.toString().orEmpty() + " %"
                    )

但如果我用资源来做：

                    perfusionIndexValue.postValue(
                        list.lastOrNull()?.perfusionIndex?.takeIf {
                            it != PerfusionIndex.NO_DATA
                        }?.value?.toString().orEmpty() + Resources.getSystem().getString(R.string.pi_percent)
                    )

我明白了

io.reactivex.rxjava3.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | android.content.res.Resources$NotFoundException: String resource ID #0x7f0e0081

我如何设置像“15%”这样的文本

我正在改编来自三词地址的一些示例代码，以便通过 Java SDK 访问他们的 API。它使用 RXJava。

示例代码是：

Observable.fromCallable(() -> wrapper.convertTo3wa(new Coordinates(51.2423, -0.12423)).execute())
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(result -> {
            if (result.isSuccessful()) {
                Log.i("MainActivity", String.format("3 word address: %s", result.getWords()));
            } else {
                Log.e("MainActivity", result.getError().getMessage());
            }
        });

首先。这会在构建时给出弃用警告以及 IDE 警告 ( Result of 'Observable.subscribe()' is ignored)。

为了解决第一个问题，我Disposable myDisposable = 在Observable. 它是否正确？（添加位置见下文）

接下来，我需要添加超时，以便在请求超时时可以显示警告等。为此，我已添加.timeout(5000, TimeUnit.MILLISECONDS)到构建器中。

timeout这是可行的，但s 似乎对 s 起作用的方式Observable是它们抛出异常，而我不知道如何捕获和处理该异常。

我现在拥有的是：

Disposable myDisposable = Observable.fromCallable(() -> wrapper.convertTo3wa(new Coordinates(51.2423, -0.12423)).execute())
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .timeout(5000, TimeUnit.MILLISECONDS)
        .subscribe(result -> {
            if (result.isSuccessful()) …

我有两个片段,每个片段都有发布主题,在onResume()初始化时调用onNext(true).我想结合这两个主题,并在它们都返回true时调用第三类中的某些方法.我需要使用Observable吗？我找不到合适的操作,zip不起作用,因为这些是主题.我该如何结合这些？

假设我有两个可观测量:A和B发射物品:A:1,2,3,4,5 B:2,4,6

有没有办法通过删除也是第二序列的项目来过滤第一个序列？

[edit]所需的数据流:从可观察的B(bList)加载所有项目,然后从A加载项目,根据标准过滤它们:!bList.contains(item)

我想用rxAndroid重写项目.在我的代码中,我有这个简单的部分:

public void chkLogin(String login, String password) {
    String[] myTaskParams = { login, password };
    new ChkLoginTask().execute(myTaskParams);
}

private class ChkLoginTask extends AsyncTask<String, Void, LoginStatus> {
    @Override
    protected void onPreExecute(){
        view.showLoadingSpinner();
    }

    @Override
    protected LoginStatus doInBackground(String... params) {
        return app.getAPI().checkLogin(params[0], params[1]);
    }

    @Override
    protected void onPostExecute(LoginStatus result) {
        view.hideLoadingSpinner();
        if(result == null)
            view.onConnectionError();
        else{
            view.onChkLoginSuccess(result);
        }
    }
}

在这段代码中,我只用两个参数在AsyncTask中调用我的api方法,然后根据它的结果做出反应.

现在我想用rxAndroid编写这段代码,但我坚持下去......我知道rx如何处理对UI元素的反应,但无法找到有关它如何与参数化AsyncTask一起工作的信息

在这种情况下使用RX是否正确？

每当我们声明流式api时使用GRPC

rpc heartBeat(Empty) returns (stream ServiceStatus){}

我们谷歌搜索观察者模式的简单接口StreamObserver(这是protobuf将为我们生成的)

public interface StreamObserver<V> {
  void onNext(V var1);

  void onError(Throwable var1);

  void onCompleted();
}

现在你想要做的是将它转换为实际,Observable并且只有在传递之后才能进一步使用.

override fun heartBeat(arg: Empty): Observable<ServiceStatus> {
    // we create rx java subject
    val subject = PublishSubject.create<ServiceStatus>()

    // we create grpc observer and delegate all calls to rx java
    val observer = object : StreamObserver<ServiceStatus> {

        override fun onNext(value: ServiceStatus) {
            subject.onNext(value)
        }

        override fun onError(error: Throwable) {
            subject.onError(error)
        }

        override fun onCompleted() {
            subject.onCompleted()
        } …

我想结合2个observable并在AndroidSchedules.mainThread()上执行"combine function".我添加了.observeOn(AndroidSchedules.mainThread()),但我仍然得到"java.lang.IllegalStateException:不在主线程上".

Observable<List<Post>> animateCameraAndGetPostsByProjection = Observable.combineLatest(
            mapObservableProvider.getMapReadyObservable(),
            LocationService.getUpdatedOrLastKnownLocation(this),
            (googleMap, location) -> {
                CameraUpdate cameraUpdate = CameraUpdateFactory.newLatLngZoom(new LatLng(location.getLatitude(),location.getLongitude()),15);
                googleMap.moveCamera(cameraUpdate);
                return googleMap.getProjection().getVisibleRegion();
            })
            .flatMap(vr -> new RestService().getApi().getPostsByMapProjection(vr.farLeft.latitude,vr.farLeft.longitude,vr.nearRight.latitude,vr.nearRight.longitude))
            .observeOn(AndroidSchedulers.mainThread());

rx.Observable.switchOnNext(Observable>)的行为与JavaDoc中描述的大理石图不同.

如果我尝试下面

public static void main(String[] args) throws Exception {
  Observable<Long> o1 = Observable.interval(200L, TimeUnit.MILLISECONDS);
  Observable<Long> o2 = Observable.interval(500L, TimeUnit.MILLISECONDS)
        .map(value -> value * 10);

  Observable.switchOnNext(Observable.just(o1, o2)).subscribe(System.out::println);

  Thread.sleep(2000L);
}

大理石图显示结果可能如下所示.

| time   | 0   | 200  | 400   | 500  | 1000   | 1500 |.....|
|--------|-----|------|-------|------|--------|------|-----|
| o1     |     | 0    | 1     |....  | ...    | ...  | ... |
| o2     |     |      |       | 0    |  10    | 20   | ... |
| result |     | 0 …

我在Android上使用RxJava来做一些事情,

在使用之前我总是在observable上做同样的事情:

Observable<AnyObject> observable = getSomeObservable();
// The next 2 lines are the lines that i always add them to any Observable
observable.observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.computation());

因此,Observable是通用的,可以是任何对象,如果我想在它上面添加这两行并在Statis方法中返回它,我需要使该方法也是Generic

我试图做的是通过参数传递observable,添加设置并将其返回如下:

public class UtilsObservable<T> {

    public static Observable<T> setupObservable(Observable<T> observable) {
        return observable.observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.computation());
}

我在这里遇到编译错误说:

UtilsObservable.this cannot be referenced from a static context

我的问题是:

那么这可以做到吗？通用方法采用泛型对象修改它并返回相同的类型？

我有一组并行执行的可观察对象，例如localObservable和networkObservable。如果networkObservable开始发射物品（从现在开始，我只需要这些物品），则丢弃发射的物品localObservable（也许localObservable尚未开始）。

Observable<Integer> localObservable =
            Observable.defer(() -> Observable.range(1, 10)).subscribeOn(Schedulers.io());
Observable<Integer> networkObservable =
            Observable.defer(() -> Observable.range(11, 20)).subscribeOn(Schedulers.io());