问题标签 [rx-java2]

我是新来的Rx世界，并尝试实施我AutoCompleteTextView的RxJava,RxBinding和Retrofit 2.

这是我想出的麻烦：（也许我没有以正确的方式做。）

我有一个AutoCompleteTextView，在这里我创建了我的订阅和 observables：

我的问题是当我使用方法设置我的 EditText 时setText()，它会触发下拉菜单。例如，当我从AutoCompleteTextView' 的下拉菜单中设置单词以及使用语音输入设置单词时，它会这样做。onTextChanged手动设置时有没有办法避免触发？或者我该如何解决？

当我使用以下代码时：(Observable.create())

方法 repeatUntil 不会被执行。控制台不会记录任何东西。
但是如果我使用以下代码：（Observable.just())

将执行 repeatUntil 方法。控制台将记录：

我不知道为什么 Observable.create() 不会执行 repeatUntil() 方法？

我目前只是EventBus用来将数据从 FirebaseMessagingService 传输onMessageReceived到 MainActivity ，但是随着复杂性的增加，这变得越来越棘手，如果我收到多个通知怎么办？另一方面，

由于 EventBus，数据传输需要 1 个额外的类和 2 个样板函数。

所以问题是如何使用 Rxjava 将数据从 FirebaseMessagingService 传输到 Activity ，有没有办法将整个服务转换为一些 observables？

我正在进行异步调用，10 秒后 1 分钟，这意味着将进行大约 6 次调用，但问题在于我希望它适用delay于特定condition

我想要的是将网络呼叫延迟 10 秒，除了第一次呼叫，并在 10 秒后取消网络呼叫，所以我应该在 1 分钟内有准确的 6 个呼叫。

编辑

由于场景混乱，这里重新定义了场景：

我拥有的是本地驱动程序的大量列表，我想每 10 秒后向每个人发送请求，并听取另一个订阅者检查驱动程序是否在 10 秒内没有取消它，这个过程将持续大约 1 分钟，如果一位司机取消，我应该立即向下一位司机发送请求

到目前为止编写的代码：

我正在从 Rx 1 迁移到 Rx 2 的过程中，突然在阅读帖子时我发现Single应该是用于改造调用的 observable 类型。

所以我决定试一试，在将改造调用迁移到 Rx 2 时，我还将返回值更改为Single<whatever>.

现在的问题是，我们的一些测试模拟了类似于以下内容的网络服务：

正如您在迁移调用之前看到的那样，我们过去只是通过告诉userService模拟返回一个空的 observable 来完成流。

在迁移到Single调用的“版本”时，我们不再可以使用Observable.empty()，因为调用不返回 an Observable，而是返回 a Single。

我最终做了类似的事情：

我的问题是：

1. 有没有更好的方法来做到这一点？
2. 我是否遗漏了任何我应该知道的重要信息——这样的事情实际上并不像我期望的那样。

假设我有一个处理器，它在按下按钮时发出一个布尔值，把它想象成一个切换。

我想做的是使用门的值来暂停和恢复可观察的序列，在暂停时缓冲发出的值。

我已经阅读了很多，虽然它似乎可以在其他语言的响应式扩展中实现，但 RxJava 似乎不支持它。

这是我想要实现的一个示例，它只是每秒输出一个增量值。当我按下按钮时，我希望输出停止，直到我再次按下它，这应该输出在两个按钮按下之间发出的每个项目：

有谁知道实现这样的事情的方法？

编辑我写了一篇关于如何实现这个的博客文章https://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

短篇小说：我有一个情况，我有 2 个具有单一目的的 Observable：

• 他们收到一些数据
• 他们返回修改后的数据
• 如果无法处理数据，则抛出错误

他们各自负责处理不同类型的数据。另外，我想在处理完两个数据后做一些事情。

我目前最好的实现如下，这些是我的 Observables：

我还有以下订阅：

我的问题：蓝色和红色数据被处理了两次，因为两个订阅都再次运行，我订阅了使用 Observable.zip() 创建的组合 observable。

如果不运行两次操作，我怎么能有这种行为？

我正在使用 2.0.0 版的 RxJava，似乎我无法访问AndroidSchedulers. 我无法通过 RxJava 访问主线程

例如，我有以下可运行的 java 代码。

它是关于一个生产者和几个并行消费者的。这些消费者正在运行耗时的作业，并且它们并行运行。

我想知道这个用例是否匹配 rx-java，以及如何在 rx-java 中重写它。

输出是：

我有一个Observable<Item> (A)，每次 a PublishSubject<Item> (B)有一个新的Item.

然后使用A，例如A.subscribeOn(computationScheduler).observeOn(mainThread)

目标：让来自A 的完整流在计算调度程序上完成工作，然后在主线程上使用结果。

实际：根据观察到B的位置，整个流将相应地在不同的调度程序上运行。在下面的例子中——在主线程上，即使是通过.subscribeOn()调用。

如何“强制”来自 A 的完整流在给定调度程序上完成其工作，并在另一个调度程序上分派结果？ A.compose() 不能解决问题。

实际代码：