问题标签 [rx-java]

我正在尝试使用 Vert.x 和 RxJava 压缩到 observables。我不知道我是否误解了某些东西，或者这只是某种错误。这是代码。

我想使用收到的消息中的信息发出 HTTP 请求，然后压缩两个 observables、事件总线和 HTTP 响应，以便使用来自 HTTP 响应的信息回复消息。

我没有收到我发送的消息的任何回复。

提前致谢！

所以我在玩 RX（真的很酷），我一直在转换我的 api，它访问 Android 中的 sqlite 数据库以返回 observables。

所以很自然地，我开始尝试解决的问题之一是，“如果我想进行 3 次 API 调用，获取结果，然后在它们都完成后进行一些处理，该怎么办？”

我花了一两个小时，但我最终找到了Zip 功能，它可以轻松帮助我：

伟大的！所以这很酷。

因此，当我压缩 3 个 observables 时，它们会串行运行。如果我希望它们全部同时并行运行，以便最终更快地获得结果怎么办？我玩过一些东西，甚至尝试阅读一些人们用 C# 编写的原始 RX 东西。我确信有一个简单的答案。谁能指出我正确的方向？这样做的正确方法是什么？

给定 Java 对象上的任意字段，我想创建一个 Observable 来监视该字段，并在每次字段值更改时将新结果推送到 Observer。ReactiveCocoa 有一个名为 RACObserve 的宏，它似乎正是这样做的。

我想知道如何使用 RxJava 实现类似的功能。

例如，假设我有以下简单的类：

我想创建一个Observable<State>每次更改currentState.

在 ReactiveCocoa 中，看起来我会写一些类似下面的东西（请原谅我的伪 Objective-C）：

我如何在 RxJava 中实现类似的功能？我在想我可能需要将所有对 currentState 的更改包装在一个 setter 中，但我不清楚我应该在哪里调用 Observable.create 以及如何将 currentState 的更改提供给 Observer。

在subjects包下，您有类似的类PublishSubject，BehaviorSubject我想可以将其描述为一些可用的示例Observables。

如何退订这些主题？没有unsubscribe方法并且调用onCompleted完全结束了 Observable 对吗？

我正在使用普通的旧 Java 1.6，并且对这两个库都感兴趣。

阅读文档后，我不确定差异（如果有）。谁能解释一下，或指出一些相关信息？提前致谢！

有没有办法在 rx-scala 中模拟Observable一个BehaviorSubject（但没有接口）？Observer即使其成为具有内存的 Observable，以便它可以有多个订阅，并且在每个新订阅上，它都会产生最后一个发出的值？

Observable.publish()完成了一半的工作，但它不会发出最后一个值。Observable.cache.publish()另一方面，重播所有值 - 我需要类似的东西，但它只重播最后一个发出的值，以处理无限流。

Rx-Java 解决方案也被接受，尽管首选原生 Scala 形式！

我希望我的 api 上的一个方法返回 Observable<Observable<Object>> 但我希望该方法中的代码在所有包含的 Observables 完成后知道，以便它可以关闭某些东西。做这个的最好方式是什么？

更明确地说，我正在完成这个方法：

我正在为我的应用程序建立网络。所以我决定试试 Square 的Retrofit。我看到他们支持简单Callback

和 RxJava 的Observable

乍一看，两者看起来非常相似，但是当它开始实施时，它变得很有趣......

虽然使用简单的回调实现看起来与此类似：

这非常简单明了。并且Observable很快就会变得冗长且相当复杂。

事实并非如此。你仍然需要做这样的事情：

我在这里错过了什么吗？或者这是使用Observables 的错误案例？什么时候/应该更喜欢Observable简单的回调？

更新

正如@Niels 在他的回答或 Jake Wharton 的示例项目U2020中所示，使用改造比上面的示例简单得多。但本质上问题是一样的——什么时候应该使用一种方式或另一种方式？

I am new to Scala, and just start learning to use fully non-blocking nature of Scala API

I need to execute function after 2 tasks finish, curenly I do this following way:

Is there a more straight forward way to do this?

I am using rxjava in my Android app to handle network requests asynchronously. Now I would like to retry a failed network request only after a certain time has passed.

Is there any way to use retry() on an Observable but to retry only after a certain delay?

Is there a way to let the Observable know that is is currently being retried (as opposed to tried for the first time)?

I had a look at debounce()/throttleWithTimeout() but they seem to be doing something different.

Edit:

I think I found one way to do it, but I'd be interested in either confirmation that this is the correct way to do it or for other, better ways.

What I am doing is this: In the call() method of my Observable.OnSubscribe, before I call the Subscribers onError() method, I simply let the Thread sleep for the desired amount of time. So, to retry every 1000 milliseconds, I do something like this:

Since this method is running on an IO thread anyway it does not block the UI. The only problem I can see is that even the first error is reported with delay so the delay is there even if there's no retry(). I'd like it better if the delay wasn't applied after an error but instead before a retry (but not before the first try, obviously).