消费流，转型，然后交给其他消费者（没有州）的方法

Question

新的Rx;试图弄清楚。

我正在通过事件从加速度计获取数据，将数据调整为我自己的格式，然后通过反应式扩展向其他消费者提供数据流。

我的方法部分是在调用一个重要的构造函数。有人可以看看这个，并帮助我理解我应该做什么不同吗？

希望，这段代码就足够了：

public class accel_raw_producer : IObservable<AccelerometerFrame_raw> 
{ 
    private static Spatial spatial = null; 
    private IObservable<EventPattern<SpatialDataEventArgs>> spatialEvents; 

    public accel_raw_producer() 
    { 
     spatial = new Spatial(); 
     spatial.close(); 
     spatialEvents = System.Reactive.Linq.Observable.FromEventPattern 
     <SpatialDataEventHandler, SpatialDataEventArgs>(
     handler => handler.Invoke, 
     h => spatial.SpatialData += h, 
     h => spatial.SpatialData -= h); 
     subscription = spatialEvents.Subscribe(); 
     spatial.open(-1); 
    } 


    public IDisposable Subscribe(IObserver<AccelerometerFrame_raw> observer) 
    { 
     return 
     (from evt in spatialEvents 
     let e = evt.EventArgs 
     select new AccelerometerFrame_raw 
     (
      e.spatialData[0].Acceleration[0], 
      e.spatialData[0].Acceleration[1], 
      e.spatialData[0].Acceleration[2], 
      e.spatialData[0].AngularRate[0], 
      e.spatialData[0].AngularRate[1], 
      e.spatialData[0].AngularRate[2] 
     )).Subscribe(); 
    } 
} 

public consumerClass :IObserver<AccelerometerFrame_raw> 
{ 
    accel_raw_producer accelStream; 
    IDisposable Unsubscriber; 

    public consumerClass() 
    { 
     accelStream = new accel_raw_producer(); 
     Unsubscriber = accelStream.Subscribe(this); 
    } 

    public void OnCompleted() 
    { 
     throw new NotImplementedException(); 
    } 

    public void OnError(Exception error) 
    { 
     throw new NotImplementedException(); 
    } 

    public void OnNext(AccelerometerFrame_raw accelFrame) 
    { 
     if (null == accelFrame) return; 
     AccelX = accelFrame.Acceleration.X; 
     AccelY = accelFrame.Acceleration.Y; 
     AccelZ = accelFrame.Acceleration.Z; 
     GyroX = accelFrame.Rotation.X; 
     GyroY = accelFrame.Rotation.Y; 
     GyroZ = accelFrame.Rotation.Z; 
    }  
} 

当我设置在AccelerometerFrame_raw断点，它就会被触发，但值不会传播到消费者。所以这种方法中的某些东西需要有所不同。

Answer 1

我不确定这里是否遗漏了一些细节，但是您似乎没有使用观察者输入参数，而订阅也没有对onNext值做任何事情。就个人而言，我会考虑这个摆动周围，只是传递出的IObservable，让来电者订阅它自己：

public IObservable<AccelerometerFrame_raw> AccelerometerFrames() 
{ 
    return 
     from evt in spatialEvents 
     let e = evt.EventArgs 
     select new AccelerometerFrame_raw 
     (
      e.spatialData[0].Acceleration[0], 
      e.spatialData[0].Acceleration[1], 
      e.spatialData[0].Acceleration[2], 
      e.spatialData[0].AngularRate[0], 
      e.spatialData[0].AngularRate[1], 
      e.spatialData[0].AngularRate[2] 
     ); 
}