RxCPP的行为不同于Rx.Net

Question

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

这里有两个程序，一个在Rx.Net中，另一个在RxCPP中。他们应该输出相同的指纹，但他们没有。

该程序从鼠标流中获取点，并计算出这些点之间的三角洲流。

鼠标是一个流的点，每一笔-从下到上按压是一条流。老鼠一个接一个地给出这样的流。

在这些测试中，预期的输出是：

德尔塔0是: 0,0

德尔塔1号是: 5,0

德尔塔2号是: 0,5

德尔塔3号是: 2,3

这就是Rx.Net输出的内容。

Rx.Cpp输出仅在第一行：Delta no 0是: 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

感谢@Kirk Shoop at rxcpp github :-)

这是一种HOTvCOLD行为。

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

如果您反转combine_latest，冷源和热源将起作用。

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