问题标签 [rx-py]

我是 ReactiveX 的新手，我有一个问题。如何在 ReactiveX 中保存数据。例如。我有这个代码。

为了正确工作，我需要知道价格的先前值。如何在 ReactiveX 中保存以前的价格。

我有一个处理 web 请求的 Observable，我想在不同的流中处理成功或失败，这与这个 example非常相似。我的脚本和示例之间的主要区别是我不想合并流然后订阅。我将 RxPY 1.6.1 与 Python 2.7 一起使用。

当请求失败时，脚本failure!会按预期打印。但是，当响应正常时，脚本不会打印success!. 有趣的是，当您切换订阅的顺序时，success!确实会打印出来，而failure!永远不会到达。

我想可能request无法多播，所以我尝试添加publish()到requestobservable 并connect()在创建订阅后调用。这没有帮助（所以我把它排除在上面的最小示例之外）。

我错过了什么？

我正在尝试实现拖放功能。我的 UI 布局如下所示：

框 1、2、3 和 4 是图像，当用户选择时，框周围会有一个白色边框。此选择操作使用 执行left press。

方框 5 显示了所选图像的一些基本信息。

框 1、2、3 和 4 中的图像可以拖放到框 5 中。成功拖放后，丢弃图像的缩略图也会显示在框 5 中。

执行使用RxPy. 我的拖放代码如下所示：

我start_drag在光标移动的任何地方都实现了一个缩略图。我已将这部分代码省略了，因为它不是我面临的问题的重点。

我遇到的问题是这个。如果我要选择一个图像，无论我做了多少mouse move，start_drag都会被解雇并创建缩略图。如果mouse move事件发生在上一次之后的某个时间范围之后left press，则不应将其视为拖动事件并且不会触发start_drag。相反，“即时” left press+mouse move事件将被视为拖动事件

应该将什么添加到我的事件流中，以便我能够确定left press和mouse move事件之间的持续时间？随后，如果此持续时间超过时间范围，我将忽略而不开火start_drag。还是有更好的解决方案来解决这个问题？

delay，debounce基本上要么推动或延迟流，我无法通过这些运营商实现我想要的。

我想用 RX python 设置一个简单的项目。我正在运行 python 3。

我已经设置了我的项目，并运行pip install rx了成功安装 rx 的项目。我使用pip show rxwhich print 检查了这个：

我的简单 python 脚本如下所示：

但是，我收到警告：Cannot find reference 'from_' in 'Observable | Observable'

并且在运行时，代码与方法调用一起出现在行from_上，并出现错误：TypeError: 'method' object is not subscriptable

有谁知道这里发生了什么？

我正在尝试了解 RxPy 的工作原理，但出现此错误

类型对象“ObservableBase”没有属性“创建”

我正在使用 python 3.6，我的代码是

我正在使用 Python 3.7.3。

我尝试将 RxPy 从 1.6.1 (1.x) 升级到 3.0.0a3 (3.x)。

使用 RxPy 1.x 的旧代码

如果您运行代码，您可以看到实时 CPU 监控图表。

但是，在我安装了新的 RxPy 之后

使用下面的新代码，它只显示白色，没有任何动态图表。

而且该功能update_plot实际上从未运行过。任何想法？

使用 RxPy 3.x 的新代码

我希望vals在最后一行更清楚。

有没有像这样的数组解构

或其他使代码清晰的方法？谢谢

这是我想要实现的目标：

1. 创建 2 个返回异步消息的独立对象
2. 转换 Observable 中的消息

这些函数很长，因此每个对象都在一个单独的文件中。这是我的对象的基本结构：

这是RXPY的代码：

问题 1：使用此代码，我只收到来自 object1 的消息。我应该如何调整此代码以使其工作并从两个对象接收消息？

问题 2：我希望来自 object1 的消息成为另一个类的参数，比如说 Object3。然后 Object3 将监听 Object2。对于来自 Object2 的每条消息，Object3 都会做一些事情。对于来自 Object1 的每条消息，Object3 的属性都会发生变化。但是，在 Object3 使用来自 Object2 的最新消息完成之前，属性不应更改。我不知道如何用 RXPY 做到这一点。任何想法？

使用踏板更新，代码可以工作。不过，我不确定这是继续进行的好方法。我将使用适当的杀戮过程对其进行更新，并且我仍在处理问题 2

我需要创建一个 Observable 流，它定期发出异步协程的结果。

intervalRead是一个返回 Observable 的函数，并将间隔rate和异步协程函数作为参数fun，该函数需要在定义的间隔内调用。

我的第一个方法是用区间工厂方法创建一个observable，然后用map调用协程，用from_future把它包装在一个Observable中，然后获取协程返回的值。

然而我得到的输出不是协程的结果，而是 from_future 返回的 Observable，在指定的时间间隔发出

输出：<rx.core.observable.observable.Observable object at 0x033B5650>

我怎么能得到那个 Observable 返回的实际值呢？我希望 42

我的第二种方法是创建一个自定义的 observable：

但是，订阅时from_future(task)永远不会发出值，为什么会发生这种情况？

然而，如果我intervalRead这样写：

我得到了预期的结果：42. 显然这并不能解决我的问题，但它让我感到困惑，为什么它在我的第二种方法中不起作用？

最后，我尝试了第三种方法，使用该方法rx.concurrency CurrentThreadScheduler并定期安排一个动作schedule_periodic。然而，我面临着与第二种方法相同的问题。

将不胜感激任何洞察我错过了什么或任何其他建议来完成我需要的。这是我第一个使用 asyncio 和 RxPY 的项目，我只在 Angular 项目的上下文中使用 RxJS，所以欢迎任何帮助。

我发现 Reactive 很难采用。我相信我的问题是示例过于简单，因此我不确定如何构建现实世界的解决方案。我希望有人能通过帮助我解决现实世界的问题来帮助我克服困难。我想在 RxPY v3 中做到这一点。

用例是这样的。您可以通过代码、买入和卖出的方式为多只股票提供无限的股票报价流。您还有一个从零开始的头寸数量（您拥有多少股票）。每只股票的报价和数量是无限可观察的。

您还有三个触发价格：入场价、目标价和止损价。如果您的头寸为零，则入场价格有效。如果股票价格上涨到入场价，那么你就采取了买入股票的行动。这将导致您的可观察位置更改为大于零的值。

一旦您的仓位大于零，目标价和止损价就会生效。如果价格上涨至目标价格，您将采取卖出股票获利的行动。如果价格跌至止损价，您将采取亏损卖出股票的行动。无论哪种方式，您的头寸都会再次为零，因此入场价格将再次变得活跃。

这是 RxPY 的一个很好的用例吗？如果是这样，请至少给我看伪代码，如果不是真正的代码，来实现这个？我不知道如何解决这个问题。最好是在一个大表达式中完成，还是在某种程度上是多个表达式的交互？动作是否总是停留在一个永无止境的可观察链中，或者这些东西会随着状态的变化而生死存亡？

非常感谢任何输入。