FCC/Framework/Edge.Core/Processor/IDeviceHandler.cs

using Edge.Core.Parser;
using Edge.Core.Processor.Communicator;
using System;
using System.IO;
using System.Threading.Tasks;

namespace Edge.Core.Processor
{
    public interface IDeviceHandler<TRaw, TMessage> where TMessage : MessageBase
    {
        /// <summary>
        /// The communicator has NOT started yet at this stage.
        /// </summary>
        /// <param name="context"></param>
        void Init(IContext<TRaw, TMessage> context);

        /// <summary>
        /// will be called every time that a message incoming from communicator.
        /// </summary>
        /// <param name="context"></param>
        Task Process(IContext<TRaw, TMessage> context);

        /// <summary>
        /// The method will be called from user for a generic test purpose, should implemented this method 
        /// to cover most real life time logic to give user a meaningful and confidence result for how this processor would work 
        /// when run for real.
        /// The method was guranteed to be called after DeviceHandler.Init, and Communicator.Start
        /// </summary>
        /// <param name="parameters"></param>
        /// <returns>throw exception to indicates the test failed, otherwise, return a Completed task.</returns>
        Task Test(params object[] parameters) { throw new NotImplementedException("暂不支持测试"); }
    }

    public abstract class TestableActivePollingDeviceHandler<TRaw, TMessage> : IDeviceHandler<TRaw, TMessage> where TMessage : MessageBase
    {
        private IContext<TRaw, TMessage> context;
        public virtual void Init(IContext<TRaw, TMessage> context)
        {
            this.context = context;
        }

        public Task Test(params object[] parameters)
        {
            var errorMsgReadTcs = new TaskCompletionSource<CommunicatorErrorMessageReadEventArg>();
            EventHandler<CommunicatorErrorMessageReadEventArg> onErrorMessageReadEvtHandler = (s, a) =>
            {
                if (errorMsgReadTcs.Task.Status == TaskStatus.RanToCompletion
                    || errorMsgReadTcs.Task.Status == TaskStatus.Faulted 
                    || errorMsgReadTcs.Task.Status == TaskStatus.Canceled)
                    return;
                errorMsgReadTcs.SetResult(a);
            };

            var dataReceivedTcs = new TaskCompletionSource<CommunicatorEventArg<TRaw, TMessage>>();
            EventHandler<CommunicatorEventArg<TRaw, TMessage>> onDataReceivedEvtHandler = (s, a) =>
            {
                if (dataReceivedTcs.Task.Status == TaskStatus.RanToCompletion
                       || errorMsgReadTcs.Task.Status == TaskStatus.Faulted
                       || errorMsgReadTcs.Task.Status == TaskStatus.Canceled)
                    return;
                dataReceivedTcs.SetResult(a);
            };
            try
            {
                this.context.Communicator.OnErrorMessageRead += onErrorMessageReadEvtHandler;
                this.context.Communicator.OnDataReceived += onDataReceivedEvtHandler;

                var i = Task.WaitAny(new Task[] { Task.Delay(12000), errorMsgReadTcs.Task, dataReceivedTcs.Task });
                if (i == 0)
                    throw new TimeoutException("12秒内未收到任何通讯层(从设备发入至 FCC)数据, 可能线路连接问题,或者设备问题?");
                if (i == 1)
                    throw new InvalidDataException("收到了一些未识别的通讯层(从设备发入至 FCC)数据,可能线路设置问题,或者设备问题?" + Environment.NewLine + (errorMsgReadTcs.Task.Result?.Hint));
                if (i == 2)
                { }
                return Task.CompletedTask;
            }
            finally
            {
                this.context.Communicator.OnErrorMessageRead -= onErrorMessageReadEvtHandler;
                this.context.Communicator.OnDataReceived -= onDataReceivedEvtHandler;
            }
        }

        public virtual Task Process(IContext<TRaw, TMessage> context)
        {
            throw new NotImplementedException();
        }
    }
}