使用反应扩展到对观察到的流

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); 
     } 
    } 
} 

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

问题2：是否使用主题OK和我实现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); 
}