RxCPP 的行为与 Rx.Net 不同答案

【问题标题】：RxCPP behave different than Rx.NetRxCPP 的行为与 Rx.Net 不同
【发布时间】：2017-08-13 15:48:16
【问题描述】：

我正在使用 RxCPP，但很难理解它的行为。

这里有两个程序，一个在 Rx.Net 中，另一个在 RxCPP 中。他们假设输出相同的打印件，但事实并非如此。
该程序从鼠标流中获取点，并计算点之间的增量流。
鼠标是点流，每个笔划 - 从下到上按是一个流。鼠标一个接一个地给出这样的流。

在这些测试中，预期的输出是：
Delta no 0 是：0,0
1 号三角洲是：5,0
2 号三角洲是：0,5
三角洲 3 号是：2,3
这是 Rx.Net 输出的内容。
Rx.Cpp 仅输出第一行：Delta no 0 is: 0,0

有什么想法吗？

Rx.Cpp 示例：

  #include <rx.hpp>
  namespace rx = rxcpp;
  namespace rxsub = rxcpp::subjects;
  using rxob = rx::observable<>;

    struct Point
    {
        Point(int x, int y) : x(x), y(y) {}

        int x = 0, y = 0;
        Point operator-() const { return {-x, -y}; }
        Point operator+(const Point& other) const { return Point{x + other.x, y + other.y}; }
        Point operator-(const Point& other) const { return operator+(-other); }
    };

    std::ostream& operator<<(std::ostream& o, const Point& p)
    {
        return o << "(" << p.x << ", " << p.y << ")";
    }

    void TestRxCPP()
    {
      using RxPoint = rx::observable<Point>;
      using Strokes = rx::observable<RxPoint>;
      using StrokesSubject = rxsub::subject<RxPoint>;

      StrokesSubject mouseSource;
      auto strokes = mouseSource.get_observable();

      auto deltaVectors = [](Strokes strokes) {
        auto deltas = strokes.flat_map([=](RxPoint stroke) {
            auto points = stroke;
            // create stream of delta vectors from start point
            auto firstPoint = points.take(1);
            auto delta =
                points.combine_latest([](Point v0, Point v1) { return v0 - v1; }, firstPoint);
            return delta;
        });

        return deltas;
      };

      auto delta = deltaVectors(strokes);
      int n = 0;
      delta.subscribe(
        [&](const Point& d) { std::cout << "Delta no. " << n++ << " is: " << d << std::endl; });

      auto testMouse = rxob::from(Point{3 + 0, 4 + 0}, Point{3 + 5, 4 + 0}, Point{3 + 0, 4 + 5}, Point{3 + 2, 4 + 3});
      mouseSource.get_subscriber().on_next(testMouse);
    }

Rx.Net 示例：

    void RxNET()
    {
        var strokesS = new Subject<IObservable<Point>>();

        Func<IObservable<IObservable<Point>>, IObservable<Point>> 
        deltaVectors = strokes =>
        {
            var deltas = strokes.SelectMany(stroke =>
            {
                var points = stroke;
                // create stream of delta vectors from start point
                var firstPoint = points.Take(1);
                var deltaP =
                    points.CombineLatest(firstPoint, (v0, v1) => new Point(v0.X - v1.X, v0.Y - v1.Y));
                return deltaP;
            });

            return deltas;
        };

        var delta = deltaVectors(strokesS);
        var n = 0;
        delta.Subscribe(d => { Console.WriteLine($"Delta no {n++} is: {d}\n"); });

        var testMouse = new List<Point>
        {
            new Point(3 + 0, 4 + 0),
            new Point(3 + 5, 4 + 0),
            new Point(3 + 0, 4 + 5),
            new Point(3 + 2, 4 + 3)
        }.ToObservable();
        strokesS.OnNext(testMouse);
    }

【问题讨论】：

您是否尝试过使用调试器或包含跟踪语句来找出这两个功能的不同之处？
调试 rxcpp 库非常困难（对我来说），它非常高端的 c++ :-(

标签： c++ rxcpp

【解决方案1】：

Thanks to @Kirk Shoop at the rxcpp github :-)
这是 HOTvCOLD 行为。

笔画是冷的并且正在共享并且只使用一个线程。 points.combine_latest(..., firstPoint) 表示在订阅firstPoint 之前发送所有积分。因此只发出最后一个增量。

如果您反转 combine_latest，COLD 和 HOT 源将起作用

auto delta =
    firstPoint.combine_latest([](Point v0, Point v1) { return v1 - v0; }, points);

【讨论】：