1

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

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

在这些测试中，预期的输出是：
Delta no 0 是：0,0
Delta no 1 是：5,0
Delta no 2 是：0,5
Delta no 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);
    }
4

1 回答 1

1

感谢 rxcpp github 上的@Kirk Shoop :-)
这是一个 HOTvCOLD 行为。

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

如果您反转 COLD 和 HOT 源将工作combine_latest

auto delta =
    firstPoint.combine_latest([](Point v0, Point v1) { return v1 - v0; }, points);
于 2017-08-14T16:24:18.653 回答