消費流，轉型，然後交給其他消費者（沒有州）的方法

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