问题标签 [reactivex]

我正在尝试通过子类化 Observable 并覆盖 lift 方法来继承 RxJS Observable 类，如此处所述。

这适用于我添加到原型中的任何运算符，但是每当我尝试实例化我的子类的新 Observable（例如，使用 MyObservable.from([1, 2, 3])）时，我都会得到父类的 Observable。

如何正确子类化 Observable，以便使用 Observable.return / Observable.from / 按预期工作？jsbin

我从github下载RxCpp，修改CMakeLists.txt如下（我使用微软修改的CMake，以便用VS2015构建跨平台项目）：

使用以下命令创建了 Project.sln 解决方案：

我稍微调整了 CMake 构建的项目，以便：

1. 启用对 C++11 的支持
2. 启用例外
3. STL 的使用：GNU STL 静态库 (gnustl_static)

我使用以下代码添加了一个简单的 cpp 文件：

项目无法编译并给出数百个错误。如果我只是尝试使用这一行进行编译：

一切正常。如果我做

智能感知似乎只是识别一些方法（不多）。我不清楚发生了什么。我被困住了。有人对尝试什么有一些建议吗？有人在为 Android 编译 RxCpp 方面取得了一些成功吗？

//更新 - 2016 年 3 月 1 日

经过一些分析，大多数错误消息不是编译错误，而是与智能感知错误有关。所以现在库编译成功。我设置了一个使用简单 c++ 函数的简单 Xamarin 应用程序，如下所示：

如果我在手机上编译和部署一切正常（甚至认为真的没有使用“反应性”代码......）。问题是当我取消注释以下行时：

然后程序编译正常（没有错误），生成了 libTest.so（大小几乎是注释行版本的两倍）但是一旦调用 Test2() 函数，我得到：

System.DllNotFOundException：libTest.so

我检查了 adb.exe，确实 libTest.so 存在于：/data/data/"package_name"/lib/

这是怎么回事？

//更新 - 2016 年 6 月 1 日

最后我找到了一些时间来准备一个重现问题的 repo：

https://github.com/easysoft2k15/AndroidRxCppTest.git

我的怀疑是，由于某些原因，当我包含使应用程序崩溃的行时，.so lib 的大小会增加超过一定限制，因此 Android 不再加载它。您的想法将不胜感激。

谢谢你

亚历山德罗

给定一个整数序列，我想根据一组条件将它分成不同的组：

• 条件可能会改变
• 每个元素可以属于零个、一个或多个组。这意味着这些组可以相互重叠。

例如，下面的代码将序列拆分1..12为两组 2 和 3 的倍数：

输出：

我的问题是：

1. 请注意，在上面的示例中，我们只有 1 个条件集divisors。以后怎么更新？我认为一个变化的变量可以很自然地被认为是一个热可观察的。但在这种情况下，我不知道如何Observable<Observable<Integer>>在第二个 observable 中使用 hot。
2. 如何按特定顺序组合拆分组，例如join？
3. 有没有更好的解决问题的方法？

Intro to Rx的调度和线程部分说

SubscribeOn 和 ObserveOn 的使用只能由最终订阅者调用

它还说，在 UI 应用程序中，通常是最终订阅者的表示层应该是调用这些方法的那个。

我想知道这个建议是否可靠，因为我看到了一些不方便的情况：

1. 对于初学者，我不认为表示层应该决定订阅来自数据层的 Observable 的位置。在我看来，表示层应该不知道数据是来自数据库、来自 REST API 还是来自内存。出于这个原因，数据层subscribeOn()在返回 Observable 之前调用会很方便，传递 IO 调度器或直接调度器一样方便。
2. 如果表示层从某个服务或用例（反过来从数据层获得）获取 Observable，并且该服务决定它需要在某个计算调度程序中处理流，那么表示层为什么要关心这个呢？
3. 最初来自 UI 的流呢，所以它需要在 UI 线程中订阅。然后它会被发送到一些服务做一些工作，最后回到表示层在 UI 线程中被观察。这将要求 UI 流成为subscribeOn()UI 调度程序，然后是observeOn()其他一些调度程序，最后observeOn()是 UI 调度程序。subscribeOn()在这种情况下，只能在最终订阅者中调用observeOn()意味着流只能在 UI 线程中处理。

为什么我应该牺牲我的应用程序的架构并忽略 Rx 通过仅由最终订阅者调用这两种方法来轻松切换线程的能力有什么好的理由？

我有以下场景，我需要在 RxJava 中对其进行转换：

1. 从本地数据库获取 30 个项目
2. 使用这 30 个条目调用 api（一个请求）
3. 从本地数据库中删除项目（这需要访问项目列表）
4. 循环（？）我想重复 1-3 直到 db 为空

到目前为止我所做的看起来像这样：

问题：

1. 发布正确的调用方式还是我应该在何时/和/然后使用？
2. 如何将模型列表从一个可观察对象传递到另一个对象？

谢谢

上下文：为了使用 JSON 树生成 swift 代码，我以预排序格式遍历树，向下到达所有叶节点，有效地将树扁平化为数组。

描述：使用 Rx.Observable.generate() 从数组元素创建一个可观察的流，但是在过滤时我得到了一些奇怪的结果。我在下面提供了一个简化的示例：

例子：

问题：为什么我得到的输出结果是 1,3,5,7,9 而不是 0,1,2,3,4,5,6,7,8,9？

我应用合并的方式似乎并不重要......当反转时我也会得到输出 0,2,4,6,8。

编辑，npm install & node version user3743222：感谢您的反馈，项目信息如下：

我正在实现一个字节消息反序列化器，它将在调度程序接口上调度反序列化消息并返回所有Throwable抛出的 observable，以便客户端代码可以处理错误。

这样做的方法原型的草图：

现在最近我很熟悉Subject<T, R>，它非常适合这里，例如

但是没有任何方便的方法create()可以轻松地委托给观察者和可观察者。所有具体实现都T与统一R，所以我无法使用其中之一。

所以我的具体问题是：有没有一种方法可以实例化一个合适Subject<byte[], Throwable>的委托给一个Observerand Observable？有没有其他方法可以创建这样的 aSubject而不必实现（在必须手动委托每个实现的方法的意义上）整个Subject,Observable和Observer？

使用 RxJava，我有一个源 Observable，它发出许多我希望与另一个发出相同类型的 Observable 相交的项目。在研究了许多选项之后，似乎最连贯的方式来构建事物是这样的：

这里的预期输出是，当我订阅生成的 Observable 时，会发出 12 个项目。由于 getMatches 的合同，我需要缓冲结果。

从表面上看，这似乎可行，但似乎不是最干净的方式。过滤器似乎不适用于此处，因为出于性能原因，我无法对每个项目运行相交检查。我玩弄了使用 flatMap 但 getMatches observable 完成了流，而不是来自源 observable 的完成通知。

有没有更好的方法来构建它？

编辑：澄清这种代码风格发生了什么：

这显然更干净，但是当我添加一些日志记录时（假设数据大小与原始片段相同：

我得到了 20 个“处理”日志实例，然后奇怪的是“处理之后”只有几条日志行（当我期望 12 时）。它似乎比应有的时间更早地调用了完成。也许我的结构有问题？

嗨，我开始使用 ReactiveX c#，一切听起来都不错，除了一个概念。

我不知道如何制作一个反应数组（或集合或列表）

我想要像 ObservableCollection 这样的东西，但有反应的概念。我的意思是没有一个数组成员是反应性的，但数组本身是反应性的，就像 Observable 集合一样。

如果反应式中的所有内容都只是一个流，我不明白如何使这成为可能......

所以问题是我应该使用什么接口或对象来获取反应数组？

谢谢

我目前正在使用这种模式来处理来自某个事件接收器（例如企业服务总线、队列、EventStore）的事件。这个想法是在一个列表中实例化几个 IEventWorker（参见下面的具体示例）并传递事件，例如，通过调用每个实例的 HandleEvent 方法，同时在列表上循环。这是 IEventWorkers 具体的基本代码：

也许有更好的方法来做到这一点？

PS：

这与我想要实现的目标非常接近。我不完全理解 ObserverRegistry 将如何工作以分发来自例如事件队列的所有事件。