问题标签 [publishsubject]

我正在开发一个显示 26 个（字母）按钮的应用程序。现在它们都连接到 中的同一个函数，ViewController依次调用 中的另一个函数ViewModel，最后这个函数调用publishSubject.onNext(). ViewController订阅了，这PublishSubject就是整个圈子。

我这样做的原因不是将发布主题绑定到按钮的点击，而是我必须知道发件人是谁。反正有这样做吗？如果没有，还有比我头脑中更智能的设计吗？

谢谢！

我有一个通用Event类，我想在PublishRelay充当我的公共汽车上发布。

我想让我RxBus的帖子事件成为通用类型，但总线本身不是通用的。你能告诉我我需要改变什么吗？

当我在代码下面运行时，如果我不写observeOn行，应用程序会崩溃，因为getView().showBlockLayout(isBlock);调用了一个尝试隐藏或显示布局的方法。但我试图在下面更改observeOn(AndroidSchedulers.mainThread())为subscribeOn(AndroidSchedulers.mainThread())应用程序再次崩溃！

我也对此进行测试：

出乎意料的是它起作用了并且没有崩溃！我没有在 getBlockObservable 方法中使用 subscribeOn（因为我知道我们可以设置一次）

这是我的UserStore课

这是我在 gradle 中导入 rxjava 依赖项的方式

我正在尝试构建一个演示器，它在某个时间段内计算一些事件，仅在第一次加载时显示加载，并在完成后更新 ui。因为可以通过多种方式（例如用户首选项）更新事件，所以我需要能够告诉演示者事件已更新并且必须再次刷新它们。这是我现在所拥有的：

主题是PublishSubjectInt 的。我onNext()在订阅后就这样做了，因为我希望数据在订阅后立即刷新。

上面的代码在我的单元测试中创造了奇迹，而且只有当我在连接了调试器的设备上运行它时。如果我只是运行它（没有任何调试器），它会到达该view.showLoading()部分，但永远不会provider.calculateEventsBetween(TimePeriod.aYearFrom(firstDay)因此 UI 被加载“卡住”。

有任何想法吗？

我正在尝试学习 RxJava2，并将我的 AsyncTasks 转换为 Observables。

我有以下要转换的代码。

我尝试使用 Disposables 重新创建以下内容。

对于每个输入

但是，currentTask始终为null。为什么？这是错误的方法吗？

我实现了一个名为“FilterByLatestFrom”的伪运算符作为 kotlin 的扩展函数。

我使用此运算符编写了下一个代码：

输出是：

我希望过滤器运算符将根据可观察 B 的最后一个值过滤 obsA。它适用于第一个块，但是当我使用新值添加 On-next 时，它不会改变结果（使用与原始可观察到的相同的最后一个值）。

这是 FilterByLatestFrom impl（它的设计也可以从 Java 中使用（使用 compose）：

我错过了什么？

编辑：我想我发现了问题： PublishSubject 应该是 BehaviorSubject 。并且合并函数应该是 conacat 以保证 obsC 将在 obsB 之后发出。

我需要澄清使用 RxJava2 管理特定场景的最佳方法是什么（整个应用程序结构都基于它）：

在我的应用程序中，很多人都可以在同一个文档中进行更改，因此我需要将每个更改传递给查看文档的每个人。但是这个对象非常复杂和沉重，所以我需要在最后一个人关闭它时将它从内存中删除。还有更多：该文档可以是另一个文档的子文档，因此父文档中的每个更改都必须发送给所有子文档。

到目前为止我所做的：我创建了一个管理器，因此每个文档请求都会发送给它。每当有人需要处理文档时，我会在地图中查看文档是否已打开。如果不是，我创建一个 BaseDocument 实例，它接收来自文档的数据和一个 PublishSubject 来分发事件并添加到该地图。然后我在 PublishSubject 上订阅用户的 Observer 以获取更改。每当用户需要更改某些内容时，它将更改发送到 BaseDocument，它进行更改并通过 onNext() 将新版本发送给每个人。到目前为止，一切都很好。

我的问题是我无法控制何时有人处置文档观察者，因此我无法控制何时不再需要该文档，因此我可以保留任何未保存的更改并销毁该对象。除了“hasObservers()”之外，我找不到任何订阅列表或类似的东西，我不想添加一个计时器来轮询它以在一切完成后关闭它。

我的“神奇答案”将是一个回调，当最后一个订阅者处理时调用，所以我可以打扫房子并将整个物体扔掉，但我找不到这样的东西。那么，我该如何管理订阅呢？

我在这里找到了 repeateWhen 的例子

但是如何在 rxJava 2 上编写它？没有 .asObservable() 函数，也没有示例！请需要你的帮助

我有一个Subject可观察的表示需要传递给多个订阅者的网络请求的结果。

我可以使用ReplaySubject缓冲区 1 和publish()方法。但是，网络请求只执行一次。

我想在将来的任何给定点触发 fetch 事件。如何触发新请求？

目前，我有一个Service包含 的对象，ReplaySubject并且有一个reload()触发网络请求并将结果发布到 a 的方法ReplaySubject。

是否有任何方法Observable可以“刷新”它并向所有当前订阅者提供新价值？

我正在使用 rxjava 2 并尝试使用 rxbus 传递值

rxbus 代码

我将值设置为

这是我的订阅代码

现在我在第一次调用时得到了值，但是当我用不同的值再次调用时，我没有得到回调。