BPASmartClient/BPASmartClient.Business/Plugin/OrderProxy.cs

using BPA.Message;
using BPASmartClient.EventBus;
using BPASmartClient.Helper;
using BPASmartClient.Model;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace BPASmartClient.Business
{
    /// <summary>
    /// 订单代理
    /// </summary>
    public class OrderProxy : IPlugin
    {
        //订单队列
        private ConcurrentQueue<MorkOrderPush> orders = new ConcurrentQueue<MorkOrderPush>();
        //运行标识
        private bool running = false;
        //设备管理
        private DeviceMgr deviceMgr;
        /// <summary>
        /// 初始化
        /// </summary>
        public void Initialize()
        {
            deviceMgr = Plugin.GetInstance().GetPlugin<DeviceMgr>();

            Plugin.GetInstance().GetPlugin<MQTTMgr>().SetMessageReciveHandler(delegate (IMessage orderInfo)
            {
                if (orderInfo == null) return;
                if (orderInfo is MorkOrderPush morkOrderpush)
                {
                    orders.Enqueue(morkOrderpush);
                }
            });

            ThreadManage.GetInstance().StartLong(() =>
            {
                while (running)
                {
                    while (orders.Count > 0)
                    {
                        var temp = orders.ElementAt(0);
                        var device = deviceMgr.GetDevices().FirstOrDefault(p => p.DeviceId == temp.DeviceId);
                        if (null != device)
                        {
                            if (!device.IsBusy && device.IsHealth)
                            {
                                if (orders.TryDequeue(out temp))
                                {
                                    var orderEvent = DoOrderEvent.Make(temp);
                                    orderEvent.Id = device.DeviceId;
                                    orderEvent.Publish();
                                }
                            }
                        }
                    }
                    Thread.Sleep(50);
                }
            }, "MQTT 消息处理");
        }

        /// <summary>
        /// 资源释放
        /// </summary>
        public void Dispose()
        {
            running = false;
        }
    }
}