使用RX構建傳感器監測系統

Question

有關背景信息，請參閱Merging multiple custom observables in RX。

我的場景是我有一些任意的傳感器（硬件）。我已經編寫了一些可插入模塊，可以用C＃連接到這些傳感器。他們目前每個人都使用一個線程來運行定時器上的採集程序。更大的目標是改變投票到RX，一旦我明白了！

需要監測這些傳感器的分組，因此我認爲會有一個聚合主題，其中顯示器可以訂閱特定傳感器組的更新（溫度，信號強度等），並可能對系統的行爲基於傳感器的讀數。

此外，每個傳感器將可能連接到記錄觀測記錄它們的當前狀態和監視器將連接到一個日誌觀測記錄其決定

同樣的設計模式將適用於任何新的傳感器，監視器或記錄儀我們介紹。

示例代碼如下：

using System; 
using System.Threading; 
using System.Collections.Generic; 

namespace Soln 
    { 
    class MainClass 
    { 
     public static void Main (string[] args) 
     { 
      Console.WriteLine ("Hello World!"); 
      var sensorA = new ASensor(); 
      sensorA.Start(); 

      var sensorB = new BSensor(); 
      sensorB.Start(); 

      var list = new List<ICustomEventHandler<string>>(); 
      list.Add (sensorA); 
      list.Add (sensorB); 

      var strObserver = new StringObserver (list); 
      strObserver.StartMonitor(); 
      Console.Read(); 
      sensorA.Stop(); 
      sensorB.Stop(); 
     } 
    } 

    //its a modular framework so every module implements 
    //this interface to interface to a core that loads them up etc 
    public interface IPlugin 
    { 
     bool Start(); 
     void Stop(); 
    } 

    public interface ICustomEventHandler<T> 
    { 
     event MyEventHandler<T> SomethingHappened; 
    } 
    //most sensors inherit from a base class and 
    //most create a thread to work in. 
    //The base interface also has an event that it uses to transmit 
    //notifications. The actual eventhandler is genericised so 
    //can be anything from a primitive to an actual object. Each plugin 
    //can additionally transmit multiply types but this is a basic example. 
    //hopefully once i can understand how rx works better , i can change the event handling to an IObservable interface 
    public abstract class Plugin<T>:IPlugin,ICustomEventHandler<T> 
    { 
     Thread oThread; 

     protected volatile bool _continueWorking = false; 
     #region IPlugin implementation 
     public bool Start() 
     { 
      oThread = new Thread (DoWork); 
      _continueWorking = true; 

      oThread.Start(); 
      return true; 
     } 
     protected abstract void DoWork(); 

     public void Stop() 
     { 
      _continueWorking = false; 
     } 

     protected void RaiseEvent(T eventMessage) 
     { 
      if (SomethingHappened != null) { 
       SomethingHappened (eventMessage); 
       Console.WriteLine (eventMessage); 
      } 
     } 
     #endregion 
     public event MyEventHandler<T> SomethingHappened; 
    } 

    public class ASensor:Plugin<string> 
    { 
     protected override void DoWork() 
     { 
      //can't share the code for company reasons 
      while (_continueWorking) { 
       Console.WriteLine (" A doing some work"); 
       Thread.Sleep (1000); 
       RaiseEvent ("ASensor has an event"); 
      } 
     } 

    } 
    public delegate void MyEventHandler<T>(T foo); 

    public class BSensor:Plugin<string> 
    { 
     protected override void DoWork() 
     { 
      //can't share the code for company reasons 
      while (_continueWorking) { 
       Console.WriteLine ("B doing some work"); 
       Thread.Sleep (1000); 
       RaiseEvent ("BSensor has an event"); 
      } 
     } 
    } 
    //the observer should be strongly typed and take a list of 
    //plugins to monitor. At least those are my current thoughts,happy 
    //to find a better way. There could be multiple observers all monitoring 
    //the same plugins for different purposes 
    public abstract class Observer<T> 
    { 
     protected List<ICustomEventHandler<T>> Plugins; 
     protected Observer(List<ICustomEventHandler<T>> plugins) 
     { 
      Plugins = plugins; 
     } 
     //use rx to subscribe to all events 
     public abstract void StartMonitor(); 
    } 

    public class StringObserver:Observer<string> 
    { 

     public StringObserver(List<ICustomEventHandler<string>> plugins) 
      :base(plugins) 
     { 
     } 

     //subscribe to all plugin events in list using rx merge? 
     //monitor and log to file 
     public override void StartMonitor() 
     { 
      //throw new NotImplementedException(); 
     } 
    } 
} 

感謝您閱讀

Answer 1

這不是完全正面回答，但它可能給你一些想法在您的代碼可以前往。

因爲您沒有給我們提供您的傳感器的代碼，我不能真正給你一個具體的解決方案，但是如果你要求我將當前代碼轉換爲Rx，那麼我可能會這樣做：

Func<string, IObservable<string>> generate = t => 
    Observable.Interval(TimeSpan.FromSeconds(1.0)).Select(x => t); 

var subject = new Subject<IObservable<string>>(); 

using (var subscription = subject.Merge().Subscribe(Console.WriteLine)) 
{ 
    subject.OnNext(generate("A doing some work")); 
    subject.OnNext(generate("B doing some work")); 
    Console.ReadLine(); 
} 

現在Func<string, IObservable<string>>只是有點重複去除，但沒有它，我可以複製你的代碼中的5條線路的功能（其中包括一個Console.ReadLine();）。

您能告訴我們傳感器代碼嗎？