使用 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();）。

您能告诉我们传感器代码吗？