问题标签 [rx-kotlin2]

我有这样的数据流：

我已将业务逻辑（实际上返回Futures）替换为CompletableFuture.supplyAsync. 而且，是的，这就是 Kotlin，但我猜你明白了。

当我评论“死”值（57005, 0xDEAD）时，输出为：

但是，如果该“死”值出现在流中，它将失败：

我是 RX 的新手，所以很快用谷歌搜索了一个解决方案：onExceptionResumeNext: Observable.fromFuture(it)--> Observable.fromFuture(it).onExceptionResumeNext { Observable.empty<Int>() }。但是现在我的应用程序永远挂起（在产生我期望的输出之后）。看起来流永远不会结束。

我应该Observable以某种方式“关闭”它还是什么？或者，更一般地说，在使用 RX 时这是一个好方法吗？我应该以另一种方式重新考虑吗？

使用RxJava2 RxKotlinand Room，我需要查询数据库以进行公开搜索。这意味着我搜索包含名为closedvalue的属性的狩猎false。找到狩猎后，它需要将查询切换到特定的狩猎。

对于这些查询，我有 2 种方法：

它们都返回 a List，否则在没有找到寻线时查询会卡住。

我的想法是这样的

供参考，这是我的Optional课

RxJava2 中是否已经内置了类似的方法？如果没有，您将如何实施？

这是一小段代码：

为什么onComplete不在这里打电话？我应该怎么做才能处理这段代码？因为在原始代码中我不能使用.toList()方法。

我正在使用 RxJava 2.*，我想通过使用 zip 运算符合并两个可观察的结果（一个来自改造，另一个来自房间）（请随意提出更好的建议）。

来自远程服务器的模型对象与来自房间数据库的模型对象不同。

1. 我想将远程对象映射到本地对象
2. 合并这两个结果
3. 显示结果。

我的远程 API 如下所示：

}

我的 Room DAO 查询如下所示：

我已将 Observable> 转换为 Observable> ，如下所示：

但是当我尝试像这样压缩这两个 observable 时：

我得到类型推断失败，预期类型不匹配

我有一个流ResponseMessage可以是不同的子类型。我想将流拆分为流，我可以在其自己的流中处理每种类型。

我的第一次尝试导致了我看不到的结果。

我现在的问题是：将流划分为一种特定子类型的流的惯用方法是什么？

我坚持让以下示例按预期工作，我尝试使用zipand combineLatest，如下所示，withLatestFrom但是它们都没有给出预期的输出。

我想要打印以下内容：

2 - 1

2 - 2

333 - 444

假设你想在你的 Observable 链中插入一个 Completable，比如对于每个发射的元素，都有一个 Completable 运行并阻塞直到它完成，你会选择什么选项？（这里Completable.complete()只是举个例子）

1. .flatMap { Completable.complete().andThen(Observable.just(it)) }

2. .doOnNext { Completable.complete().blockingAwait() }

3. 别的东西？

你能帮我按以下 json 分组并根据 RxKotlin 的日期在 kotlin 中返回一个 hashMap 吗？只使用 kotlin 就很容易，但对于 Rxkotlin 来说真的很困难。谢谢

val groupedTransactions = accountTransactions.transactions ?.groupBy { it.effectiveDate }

我知道如何在RxJava 2中做到这一点。

而且我知道RxKotlin如何帮助解决类似的问题。

但似乎 RxKotlin.Observables 没有这个列表重载的辅助函数，我无法弄清楚。你会怎么做？

I have the following design I'd like to create, but I'm not sure which Rx pattern matches it. The goal is more or less along the lines of a Single, but with a conditional check.

• There is one Observable<String>, and the possibility of any number of observers.
• If a request is first made, the observable will execute some network request taking in the string, then emit a callback (much like a completable/single)
• Any subsequent call with the same key will return the same result immediately
• However, if 5 minutes has passed and the same call is made, we will refetch the data as it may have expired, then emit it to any listeners. This result will be saved for another 5 minutes, and the cycle repeats.
• All data is stored based on the key sent, much like a flyweight pattern. Expiration is based off of the last request time of the specific key.

My initial thought was to just make my own class with a concurrent hashmaps. However, this will mean I have to handle a lot of the threading mechanisms myself. I feel like RxJava will be a great solution to this, but I'm not sure if such patterns exist. Does anyone have an idea?

I get that the purpose of a Single<T> is meant to only retrieve a single response, so my terms may not be correct.

The following is my attempt, which I will be updating as I go