c++ - RxCPP 的行为不同于 Rx.Net

Question

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

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

在这些测试中，预期的输出是:
0 号增量为:0,0
Delta 1 是:5,0
Delta 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);
    }

Answer 1

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

这些笔画是冷的并且正在共享并且只使用一个线程。 points.combine_latest(..., firstPoint) 意味着在订阅 firstPoint 之前发送所有点。因此只有最后一个增量被发出。

如果您反转 combine_latest

，冷源和热源将起作用

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