Aufbau eines Sensorüberwachungssystems mit RX

Question

Siehe Hintergrund Merging multiple custom observables in RX.

Mein Szenario ist, dass ich eine Reihe von beliebigen Sensoren (Hardware) habe. Ich habe einige steckbare Module geschrieben, die an diese Sensoren in C# angeschlossen werden können. Sie verwenden derzeit jeweils einen Thread, um eine Erfassungsroutine auf einem Zeitgeber auszuführen. Das größere Ziel ist es, die Umfrage zu RX zu ändern, sobald ich verstehe, wie!

Es gibt eine Anforderung, diese Sensoren in Gruppen zu überwachen, also dachte ich, es gäbe ein aggregiertes Thema, bei dem ein Monitor Aktualisierungen von einer bestimmten Gruppe von Sensoren (Temperatur, Signalstärke usw.) abonnieren und möglicherweise Änderungen vornehmen könnte das Verhalten des Systems basierend auf den Messwerten der Sensoren.

Zusätzlich jeder Sensor würde möglicherweise zu einer Protokollierung Beobachter verbinden ihrem aktuellen Zustand und der Monitor an einen Logging-Beobachter anmelden anmelden seine Entscheidungen

Das gleiche Design-Muster würde gelten für alle neuen Sensor, Monitor oder Logger würde verbinden wir stellen vor.

Beispielcode unten:

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

Vielen Dank für

Lesen

Answer 1

Dies ist nicht ganz eine direkte Antwort, aber es könnte Ihnen eine Idee geben, wo Ihr Code Kopf kann.

Da Sie uns den Code für Ihre Sensoren nicht gegeben haben, kann ich Ihnen keine konkrete Lösung geben, aber wenn Sie mich bitten, Ihren aktuellen Code in Rx umzuwandeln, würde ich wahrscheinlich so etwas tun:

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

Nun ist die Func<string, IObservable<string>> war nur ein bisschen Vervielfältigung Entfernung, aber ohne sie kann ich die Funktionalität des Codes in 5 Zeilen (die eine Console.ReadLine(); enthält) replizieren.

Können Sie uns bitte den Sensorcode zeigen?