问题标签 [rx-java2]

我正在尝试创建一个 RxJava BlockingObservable，它将每 X 毫秒发出一个变量的值，直到（条件 == true）或发生超时。

下面的代码似乎接近我想要的，但它总是发出一次然后退出。奇怪的是，我有一个takeUntil()永远不会成立的条件——我希望这个 observable 持续发射并最终超时，但事实并非如此。

我在这里错过了什么/做错了什么？

Subscriber尝试创建和订阅时，我收到如下错误消息。

构建.gradle

GooglePlaceService.java

ApiUtils.java

ObservableObservable<GooglePlacesResponse>如下。

我有一个包含对我的数据库对象的引用的单例。我想做的是将任何数据库操作限制为单个 IO 线程。

首先，我尝试了以下方法：

这种方法的问题是 Schedulers.io() 可能会为每个订阅我上面的 observable 的订阅者从其池中提供一个不同的线程（我不希望这样，我希望代码在同一个线程上执行每时每刻）。

然后我转向自己持有调度程序的方法：

由于我是 RxJava 的新手（而且它看起来有很多陷阱），我问：保持 Scheduler 对象处于活动状态有什么害处（请记住 SQLSingleton 是单例，因此 Scheduler 也将是）？

从可观察对象接收到第一个项目后，我试图取消订阅。它似乎不起作用。我究竟做错了什么？

结果似乎在无限循环中打印“Hello World”。

我们如何处理 Rxjava2 中的不同网络错误？

我们曾经使用 Rxjava 1 检查 throwable 的实例，如果它是 IOException 或 HttpException 返回，但是在 RxJava 2 中，throwable 错误的类型是GaiException。

代码片段

参考：https ://github.com/square/retrofit/issues/690 https://android.googlesource.com/platform/libcore/+/5d930ca/luni/src/main/java/android/system/GaiException.java

这是代码：

当一个或多个 observables 返回代码 = 404 时，我抛出异常

我会得到

如何解决这个问题呢？

Observable如果 an在完成之前没有发出恰好一个值，我想“抛出”自定义错误。

让我试着举一个例子：

所以我有一个 SomeClass 对象流。fitler()如果不只发出一个元素，我想发出自定义错误。

在 RxJava 1 中，使用 Observer 订阅会返回一个可以取消订阅的订阅。

在 RxJava 2 中，使用 Observer 订阅会返回 void 并且没有 Disposeable。怎么可能停止那个“订阅”？

编辑：如何处理这种情况，我们使用本身不实现的观察者Disposable，比如BehaviorSubject？就像在这个例子中一样：

当 Completable 正常完成时，我期待 flatMapCompletable 将调用给定的 Action。但是，它并没有像我想象的那样工作。这是一个简单的例子：

这是预期的行为吗？如果是这样，我如何检查 Completable 任务是否完成？Completable.complete().doFinally()?

我一直在尝试了解 RxJava 2 以及如何使用它。我还试图通过将我的代码转换为 RxJava 代码来理解 RxJava。但我似乎仍然无法理解它。如果有人能进一步解释我如何重新编码，我真的会提供帮助。

这是我试图在 RxJava 中编写的代码：

我试图将 getMyUsers() 方法变成 Observable。但我仍然对如何解决这个问题感到困惑。如果有人能指出我正确的方向，我将不胜感激。:)