问题标签 [rx-java]

在带有clojure 1.5.1的lighttable下使用rxjava（http://netflix.github.io/RxJava/javadoc/），考虑以下（使用副作用，变异（可怕！）外部收集器从monad中提取数据）

产生：

我想说

但是，惊喜

事实上，只是

产生类似的异常。为什么 Clojure 可以Observable/toObservable在表达式中找到它

但不是像这样的表达

?

我提前为这篇笔记的篇幅道歉。我花了相当多的时间使它更短，这是我能得到的最小的。

我有一个谜，会很感激你的帮助。这个谜团来自observer我在 Clojure 中编写的 rxjava 的行为，它来自于observable从在线示例中抄录的几个简单的 s。

一个可观察对象同步向其观察者的处理程序发送消息onNext，而我所谓的有原则的观察者的行为符合预期。

另一个 observable 通过 Clojure 在另一个线程上异步执行相同的操作future。完全相同的观察者不会捕获发布到其的所有事件onNext；它似乎只是在尾部丢失了随机数量的消息。

promise在donCompleted的等待到期和发送到agent收集器的所有事件的等待到期之间存在以下故意竞争。如果获胜promise，我希望false在. 如果获胜，我希望看到来自's 队列的所有消息。我不期望的一个结果是AND来自. 但是，墨菲不睡觉，而这正是我所看到的。我不知道垃圾收集是否有问题，或者 Clojure 的 STM 内部排队，或者我的愚蠢，或者其他什么。onCompletedagentagenttrueonCompletedagenttrueonCompletedagent

我在这里按照自包含形式的顺序呈现源代码，以便它可以直接通过lein repl. 有三个仪式要避免：首先，leiningen 项目文件，project.clj它声明了0.9.0对 Netflix 的 rxjava 版本的依赖：

现在，命名空间和 Clojure 需求以及 Java 导入：

最后，输出到控制台的宏：

最后，给我的观察者。我使用 anagent来收集任何 observable 发送的消息onNext。我使用 anatom来收集潜在的onError. 我使用 apromise以便onCompleted观察者外部的消费者可以等待它。

现在，这是一个同步的 observable。它将 25 条消息onNext送入观察者的喉咙，然后调用他们onCompleted的 s。

我们订阅我们的观察者这个 observable：

它按预期工作，我们在控制台上看到以下结果

这是一个异步 observable，它做的事情完全相同，只是在 afuture的线程上：

但是，令人惊讶的是，这是我们在控制台上看到的：truefor onCompleted，暗示promiseDID NOT TIME-OUT; 但只有一些异步消息。我们看到的实际消息数量因运行而异，这意味着存在一些并发现象。线索赞赏。

我想使用一些在 Clojure 中经常在https://github.com/Netflix/RxJava上构建的最新功能，但是在让 Leiningen 引用我构建的本地 .jar 时遇到了一些困难。发布到 Clojars 的 rxjava 的最后一个版本是 0.9.0，我可以在我的 projects.clj 中成功地使用以下版本

现在，我成功构建了当前的 rxjava 源代码，生成了以下文件

在我的 Clojure 项目目录中，我执行以下操作

按照此处给出的说明：https ://gist.github.com/stuartsierra/3062743 （见底部）通过http://www.pgrs.net/2011/10/30/using-local-jars-with-leiningen /和leiningen - 如何为本地 jar 添加依赖项？.

报告成功并产生以下结果：

我现在修复我的 projects.clj 文件以包含以下内容：

但是 lein deps 找不到本地仓库

project.clj 中的以下猜测也不起作用：

任何想法如何进行，请&谢谢？

这段代码是线程安全的吗？

我很好奇，因为 ArrayList 不是线程安全的数据结构。

我想像使用 Guava 的EventBus或 Otto 一样使用 RxJava，但我不知道如何让它以这种方式运行。

这是场景：假设我想在我的 Android 应用程序中有一个按钮，每次按下按钮时，我都希望 RxJava 通过我的 Observable 发出一个事件。在我看来，我必须reregister在它收到一个事件后拥有该服务，并且该活动也需要创建一个新的 observable。

就像我说

在我看来，对于每个事件，我都需要这样做，但这会创建一个新的 observable，需要再次注册。当然，我错过了一些东西。

我有一个参数化的休息调用，应该每五秒使用不同的参数执行一次：

我需要创建一个Observable<TResult>将每 5 秒轮询一次 restCall 的参数，并为 param1 提供不同的值。如果 api 调用失败，我需要得到一个错误并在 5 秒内进行下一次调用。只有在 restCall 完成（成功/错误）时才应该测量调用之间的间隔。

我目前正在使用 RxJava，但 .NET 示例也很好。

所以我对整个函数式反应范式产生了浓厚的兴趣，并试图将其中的一些实践付诸实践。但我仍在努力理解的一件事是如何在前端使用这些异步服务器端调用。因此，假设您的服务层生成了一个Observable<?>对象，该对象将由您的控制器调用。伟大的。但是现在，您如何处理来自客户端的这种异步结果？通常这是你会阻塞结果的地方，比如来自一个Future<?>对象，然后构造你的结果对象并将其发送回调用客户端。但重点是不要阻塞，那么你现在用你的 Observable 做什么呢？一旦结果可用，您是否必须使用网络套接字来“推送”结果？

我希望这个问题是有道理的。只要调用和结果消费者存在于服务器端，所有这些都是有意义的。但是在 web 应用程序中，客户端在调用控制器后期望结果，我不明白如何在这里保持异步、反应范式。任何帮助都非常感谢您理解这一点。

我正在尝试在 Android 上使用 rx-java 来执行几个连续的 http 请求，每个请求都依赖于前一个的响应。这不太适合 map() / doFinall() 模型，所以我不确定在不进入“回调地狱”以及编写简洁代码的情况下最好的方法是什么。

更具体地说：

1. 做 http GET "/x"
2. 如果 (2) 成功，则执行 http GET "/y"
3. 对 GET /y 的结果进行计算

关于如何解决这个问题的任何建议？

当我想在创建 Observable 后添加元素时，如何实现场景，可以完成吗？在观察者模式中，我只会触发事件左右。你有什么想法吗？

我正在尝试学习 RxScala。我正在使用Observable.interval(1 second).buffer(10 seconds)并订阅以打印值，但无法正确取消订阅。在我取消订阅后，它会继续运行，但结果为空。

我的代码：

它可以很好地打印结果：

一段时间后，我取消订阅：

我仍然得到打印，但缓冲区为空：

我认为这意味着我的订阅仍在运行，但接收到空缓冲区，我该如何停止订阅？