标签: rx-java2

我想将互联网上的汽车列表加载到Widget中的ListView中.当我在onDatasetChanged方法中请求汽车时,它不会更新列表.然而,当我手动点击我的刷新按钮时,它会刷新它.可能是什么问题呢？为了提供尽可能多的信息,我将完整的源代码放入ZIP文件中,该文件可以解压缩并在Android Studio中打开.我也将它嵌入下面.注意:我使用RxJava作为异步请求,代码是用Kotlin编写的,尽管这个问题不是每个语言都绑定的.

CarsWidgetProvider

class CarsWidgetProvider : AppWidgetProvider() {
    override fun onUpdate(context: Context, appWidgetManager: AppWidgetManager, appWidgetIds: IntArray) {
        Log.d("CAR_LOG", "Updating")
        val widgets = appWidgetIds.size

        for (i in 0 until widgets) {
            Log.d("CAR_LOG", "Updating number $i")
            val appWidgetId = appWidgetIds[i]

            // Get the layout
            val views = RemoteViews(context.packageName, R.layout.widget_layout)
            val otherIntent = Intent(context, ListViewWidgetService::class.java)
            views.setRemoteAdapter(R.id.cars_list, otherIntent)

            // Set an onclick listener for the action and the refresh button
            views.setPendingIntentTemplate(R.id.cars_list, getPendingSelfIntent(context, "action"))
            views.setOnClickPendingIntent(R.id.refresh, getPendingSelfIntent(context, "refresh"))

            // Tell the AppWidgetManager …

我正在使用Room进行数据库管理，而在处理实时数据时该使用什么却感到困惑。现在，要管理我正在使用的实时数据，我对此Flowable非常满意。我感到困惑的是，我也可以使用LiveData它来执行相同的操作。

为了提供一些背景信息，这是我查询数据和更新视图的方式。

流动性

addDisposable(userDao().getUsersFlowable()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(users -> userAdapter.setUsers(users)));

实时数据

userDao().getUsersLiveData()
    .observe(this, users -> {
        userAdapter.setUsers(users)
    })

我不太熟悉LiveData，但是就我的研究而言，它是一个观察者模式，它也知道生命周期，这意味着我将停止通知UI是否未处于活动状态。就是说，正如您在我的Flowable代码中看到的那样，我将其添加到其中CompositeDisposable，并将在我的onDestroy()方法中进行处理。因此，我看不出为什么我应该使用LiveData什么时候可以用进行管理RxJava，因为为了方便起见，该操作有很多运算符。

所以我应该何时使用，LiveData以及何时RxJava使用Room。反映给定方案的答案非常受赞赏，但也欢迎其他用例。

我遵循了何时在Android中使用RxJava以及何时从Android Architectural Components中使用LiveData？，但范围太广，我无法具体回答我的情况

我正在使用RxJava 2& Retrofit 2（https://github.com/JakeWharton/retrofit2-rxjava2-adapter），我想知道如何处理无响应（204）类型。在 rxjava1 中，我正在使用Observable<Void>，但 rxjava2 不再允许它（https://github.com/ReactiveX/RxJava/wiki/What's- Different-in-2.0 -> Nulls ）

现在，我已经绕过自定义类型（我称之为 NoContent）的 Json 解析，但我想知道是否有更好的方法。

编辑：

public class NoContent {
    public static class GsonTypeAdapter extends TypeAdapter<NoContent> {

        @Override
        public void write(JsonWriter out, NoContent value) throws IOException {
           out.nullValue();
        }

        @Override
        public NoContent read(JsonReader in) throws IOException {
           return new NoContent();
        }
    }
}

我正在尝试创建Flowable一个包装Iterable. 我Iterable定期将元素推送到我的元素，但似乎完成事件是隐式的。我不知道如何表示处理已完成。例如在我的代码中：

    // note that this code is written in Kotlin
    val iterable = LinkedBlockingQueue<Int>()
    iterable.addAll(listOf(1, 2, 3))

    val flowable = Flowable.fromIterable(iterable)
            .subscribeOn(Schedulers.computation())
            .observeOn(Schedulers.computation())

    flowable.subscribe(::println, {it.printStackTrace()}, {println("completed")})

    iterable.add(4)

    Thread.sleep(1000)

    iterable.add(5)

    Thread.sleep(1000)

这打印：

1 2 3 4 已完成

我检查了接口的来源Flowable，但似乎我无法Flowable明确表示 a 已完成。我怎样才能这样做呢？在我的程序中，我发布的事件之间有一些延迟，我想明确何时到达complete事件流。

澄清：我有一个长时间运行的进程，它会发出事件。我将它们收集在一个队列中，并公开一个方法，该方法返回一个围绕我的队列的 Flowable。问题是，当我创建 Flowable 时，队列中可能已经有元素了。我只会处理事件一次，并且我知道事件流何时停止，因此我知道何时需要完成 Flowable。

搜索是我的应用程序的核心，我需要它才能正常工作。现在我有一个SearchView。我需要内联显示结果，因此我使用此代码。

    searchView.setOnQueryTextListener(new SearchView.OnQueryTextListener() {
        @Override
        public boolean onQueryTextSubmit(String query) {
            srl.setVisibility(View.GONE);
            return false;
        }

        @Override
        public boolean onQueryTextChange(String query) {
            currentQuery = query;
            if (query.length()>= 3) {
                searchFor(currentQuery);
            } else {
                srl.setVisibility(View.GONE);
            }
            return false;
        }
    });

问题可能很明显。因为我使用 firebase，我的searchFor()函数相当重，我不需要对每个字母都执行它。这不仅会破坏用户体验，如果您写下较长的单词，有时还会使我的应用程序崩溃。

我想要的是当用户停止打字时进行搜索。我想我需要一个处理程序，将其延迟一秒，然后每次按下字母键时取消该处理程序并设置一个新的处理程序。这在理论上是有道理的。我只是无法自己为 searchView 实现这一点。

帮助将不胜感激！

在我的 Android 应用程序中，我使用 Retrofit 2.2 和 RxJava 2 进行网络调用。当设备连接到网络时一切正常。当我禁用移动和 WiFi 网络时，应用程序崩溃。奇怪的是，我可以正确处理错误，但应用程序还是崩溃了。

这是Retrofit的API接口：

public interface BackendApi {

 @GET("v1/users/{uuid}")
 Observable<User> getUser(@Path("uuid") String userUuid);

}

以下方法返回一个将在我的视图模型中订阅的 Observable。

@Override
public Observable<User> getUser(@NonNull final UUID id) {
    return api.getUser(id.toString()).map(user -> mapUser(user));
}

在视图模型中，我订阅了 Observable：

compositeDisposable.add(userService.getUser(userId)
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(this::onGetUser, this::onGetUserError));

下面几行显示了错误处理方法

 private void onGetUserError(final Throwable throwable) {
    log.error("onGetUserError, getting user failed because of {}", 
 throwable.getCause());
    if (isViewAttached()) {
        getView().showUserLoginOrRegisterError();
    }
 }

问题是，当我断开设备与移动和 WiFi 连接的连接时， onGetUserError() 方法会被正确调用，但无论如何，存在未处理的异常导致我的应用程序崩溃。

这是完整的堆栈跟踪：

https://gist.github.com/anonymous/859ab3811a3ca5c63998a40b877c41bb

我不知道如何摆脱这些崩溃！

问候马丁

----------更新 03.04.2017--------- 还有另一个谜团...在 Samsung Galaxy (S5) …

我遇到了一个问题，我的可观察对象在 IO 线程上订阅并在 android 主（UI）线程上观察，但运算doFinally符在 IO 线程上运行，并且需要在 UI 线程上运行。

用例与这篇媒体文章几乎完全相同。

我本质上是想显示ProgressBarObservable 被订阅的时间，并隐藏ProgressBarObservable 终止或完成的时间。

我收到的错误是：java.lang.IllegalStateException: 当前线程必须有循环器！

任何人都可以帮我将doFinally操作移回具有循环程序的 UI 线程吗？或者我遗漏了其他一些信息？

编辑 用例工作流程是：

-> 启动活动

-> 初始化

-> 执行可观察流

-> 开始新的 Activity 并完成当前的 Activity

-> 新活动

-> 开始原始活动并完成

-> 重复初始化

非常感谢。

细节：

• RxJava 2.0.7
• RxAndroid 2.0.1
• Android sdk 最低 14 和目标 25

示例代码

listUseCase.execute(null)
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(@NonNull Disposable disposable) throws Exception {
                    getView().showLoading(true);
                }
            })
            .doFinally(new Action() {
                @Override …

我是 rxjava/rxkotlin/rxandroid 的初学者。

我需要按顺序处理三个不同的异步调用。问题是第一步返回 a Single<LocationResult>，第二步返回 a Completable，第三步返回 a Completable。

（单个 -> 可完成 -> 可完成）

现在的问题是最后一个Completable取决于第一个的数据Single

我当前的解决方案：

我认为这是一个糟糕的解决方案，但我不知道如何正确地做到这一点。

val ft = FenceTransaction(applicationContext, apiClient)
        stream
            .flatMap { locationResult ->
                ft.removeAll()
                return@flatMap ft.commit().toSingle({ return@toSingle locationResult })
            }
            .flatMapCompletable {
                ft.recycle()
                ft.setZone(it.location.longitude, it.location.latitude, ZONE_RADIUS)
                val dots = DotFilter().getFilteredDots()
                for (dot in dots) {
                    ft.addDot(dot)
                }
                return@flatMapCompletable ft.commit()
            }
            .subscribeBy(
                onComplete = {
                    "transaction complete".logi(this)
                },
                onError = {
                    "transaction error".logi(this)
                })

这种方法是正确的方法吗？

我应该如何处置Completeables？一般我应该什么时候处置Observables？

我对互联网、数据库进行了一系列调用，结果我向用户显示收集到的信息。现在我有非常难看的三层嵌套 RxJava 流。我真的想让它变得流畅且易于阅读，但我真的很难坚持下去。

我已经阅读了有关 Map、flatMap、zip 等的所有内容。不能让这些东西一起工作。

代码：进行api调用。接收到的信息放入数据库中，在第一个流的 onSuccess 方法中订阅另一个流，并在第二个流的 onSuccess 方法中从数据库接收到的信息最终显示给用户。

达特·弗兰肯斯坦：

disposables.add(modelManager.apiCall()
                .subscribeOn(Schedulers.io())
                .observeOn(mainThread)
                .subscribeWith(new DisposableSingleObserver {

                  public void onSuccess(ApiResponse apiResponse) {

                        modelManager.storeInDatabase(apiResponse)
                       //level 1 nested stream:
                        disposables.add(modelManager.loadFromDatabas()
                                  .subscribeOn(Schedulers.io())
                                  .observeOn(mainThread)
                                  .subscribeWith(new DisposableSingleObserver{
                                    public void onSuccess(Data data) {
                                        view.showData(data);
                                    }
                                    public void onError(Throwable e) {
                                    }
                                }));
                    }
                    @Override
                    public void onError(Throwable e) {
                    }
                }));
    }

我想限制/testAPI 调用每 3 秒调用一次，例如：

2021-09-21 14:09:19.920 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.031 V/OkHttp: <-- 200 https://xxx/test (109ms)
2021-09-21 14:09:20.038 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.136 V/OkHttp: <-- 200 https://xxx/test (96ms)
2021-09-21 14:09:20.146 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.315 V/OkHttp: <-- 200 https://xxx/test (168ms)
2021-09-21 14:09:20.325 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.499 V/OkHttp: <-- 200 https://xxx/test (172ms)
2021-09-21 14:09:20.514 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.615 V/OkHttp: <-- 200 https://xxx/test (100ms)
2021-09-21 14:09:20.628 V/OkHttp: --> GET https://xxx/test
2021-09-21 14:09:20.721 V/OkHttp: …