使用响应式扩展过滤可观察流中的“重大变化”

Question

我有一个GeoLocationProvider（它实现了IObservable<System.Device.Location.GeoCoordinate>，它每 x 毫秒输出一次当前位置。

现在我想使用 RX 读取所有这些 GPS 坐标，并且仅在位置变化重大时（例如 - 行驶距离 > 10 米）通知订阅者。

我的“主要”代码

// [...]
IObservable<GeoCoordinate> locationProvider = new GeoLocationProvider();
LocationFeed locationFeed = new LocationFeed(locationProvider);

// register any interested observers on the locationFeed.
ConsoleLocationReporter c1 = new ConsoleLocationReporter("reporter0001");
locationFeed.Subscribe(c1);

我的 LocationFeed 实现如下所示：

using System;
using System.Device.Location;
using System.Reactive.Subjects;

namespace My.Namespace.Movement
{
    public class LocationFeed : ISubject<GeoCoordinate>, IDisposable
    {
        private readonly IDisposable _subscription;
        private readonly Subject<GeoCoordinate> _subject;

        public LocationFeed(IObservable<GeoCoordinate> observableSource)
        {
            _subject = new Subject<GeoCoordinate>();
            _subscription = observableSource.Subscribe(_subject); // TODO: Add logic to filter to only significant movement changes (> 10m)
        }

        public void Dispose()
        {
            _subscription?.Dispose();
            _subject?.Dispose();
        }

        public void OnNext(GeoCoordinate value)
        {
            _subject.OnNext(value);
        }

        public void OnError(Exception error)
        {
            _subject.OnError(error);
        }

        public void OnCompleted()
        {
            _subject.OnCompleted();
        }

        public IDisposable Subscribe(IObserver<GeoCoordinate> observer)
        {
            return _subject.Subscribe(observer);
        }
    }
}

问题一： GeoCoordinate 提供了一个方法c1.DistanceTo(c2) 来计算两个坐标之间的距离。如果阈值与上次推送的阈值相比大于 x，我只想报告（发布）新的地理坐标。我如何做到这一点？

问题2：主题的使用是否正常，以及我实现ISubject的方式？我不想在我的“主”代码中添加所有连线并将其全部移到单独的类中。

Answer 1

我强烈建议不要实现ISubject<T>（或者就此而言IObservable<T> 或IObserver<T>）。而是尝试组合现有的工厂和类型，然后将它们公开为“具有”关系而不是“是”关系。

如您所见，您的LocationFeed 纯粹是observableSource 参数的包装，因此似乎无法解决任何问题。我建议删除它。

关于您发布的问题，一种解决方案是使用大小为 2 且步长为 1 的缓冲区。

IObservable<GeoCoordinate> locationProvider = new GeoLocationProvider();

locationProvider
    .Buffer(2,1)
    .Where(buffer=>buffer[0].DistanceTo(buffer[1]) > 10)
    .Select(buffer=>buffer[1])
    .Subscribe(
        pos => Console.WriteLine(pos),
        ex => { },
        () => {});

或者你可以使用Scan

IObservable<GeoCoordinate> locationProvider = new GeoLocationProvider();

locationProvider
    .Scan(Tuple.Create(GeoCoordinate.Zero,GeoCoordinate.Zero), (acc, cur)=>Tuple.Create(acc.Item2, cur))
    .Where(pair=>pair.Item1.DistanceTo(pair.Item2) > 10)
    .Select(pair=>pair.Item2)
    .Subscribe(
        pos => Console.WriteLine(pos),
        ex => { },
        () => {});

我不确定您对产生的第一个值有什么要求。是否应该发布？

编辑： 这是一个经过测试的解决方案（使用Point 类型），当发生“重大”更改时，它将推送Unit。如果这不是您想要的，那么您应该可以通过修补来获得您真正想要的

void Main()
{
    var zero = new System.Drawing.Point(0,0);
    var fenceDistance = 10;

    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        ReactiveTest.OnNext(1, new System.Drawing.Point(0,0)),
        ReactiveTest.OnNext(2, new System.Drawing.Point(0,9)),  //Not far enough
        ReactiveTest.OnNext(3, new System.Drawing.Point(0,10)), //Touches the fence
        ReactiveTest.OnNext(4, new System.Drawing.Point(0,15)), //Not far enough
        ReactiveTest.OnNext(5, new System.Drawing.Point(0,40))  //Breaches the fence        
        );

    var observer = scheduler.CreateObserver<Unit>();

    source
        .Scan(Tuple.Create(zero, zero), (acc, cur) =>
        {
            if (DistanceBetween(acc.Item1, cur) >= fenceDistance)
            {
                return Tuple.Create(cur, cur);
            }
            else
            {
                return Tuple.Create(acc.Item1, cur);
            }
        })
        .Where(pair => pair.Item1 == pair.Item2)
        .Select(pair => Unit.Default)
        .Subscribe(observer);


    scheduler.Start();

    ReactiveAssert.AreElementsEqual(new[] {
        ReactiveTest.OnNext(1, Unit.Default),
        ReactiveTest.OnNext(3, Unit.Default),
        ReactiveTest.OnNext(5, Unit.Default)
    },observer.Messages);

}

// Define other methods and classes here
public static double DistanceBetween(System.Drawing.Point a, System.Drawing.Point b)
{
    var xDelta = a.X -b.X;
    var yDelta = a.Y - b.Y;

    var distanceSqr = (xDelta * xDelta) + (yDelta * yDelta);
    return Math.Sqrt(distanceSqr);
}