#pragma once #define LOGICDEVICE_API #include #include "Definitions.h" #include "ResDataObject.h" //#include "logger.h" #include "DPC.h" #include "CcosLock.h" #include #include "CcosFileHandle.h" #include "CcosThread.h" #include "MsgQueue.h" #include #include // 添加Linux时间头文件 #include // 添加Linux线程头文件 #include // 添加unistd头文件 #include // 添加信号量头文件 // 添加 GUID 类型定义(如果未定义) #ifndef GUID_DEFINED #define GUID_DEFINED typedef struct _GUID { unsigned long Data1; unsigned short Data2; unsigned short Data3; unsigned char Data4[8]; } GUID; #endif typedef enum _SyslogLevel { Syslog_Debug = 0, Syslog_Information = 1, Syslog_Warning = 2, Syslog_Recoverable = 3, Syslog_Error = 4, Syslog_Fatal = 5, Syslog_Telegram = 6, Syslog_XReset = 7, Syslog_MAX }SYSLOGLEVEL; #define SYSLOG_DEBUG(Code,format,...) SystemLog(Syslog_Debug,Code,format,__VA_ARGS__) #define SYSLOG_INFO(Code,format,...) SystemLog(Syslog_Information,Code,format,__VA_ARGS__) #define SYSLOG_WARNING(Code,format,...) SystemLog(Syslog_Warning,Code,format,__VA_ARGS__) #define SYSLOG_ERROR(Code,format,...) SystemLog(Syslog_Error,Code,format,__VA_ARGS__) #define SYSLOG_FATAL(Code,format,...) SystemLog(Syslog_Fatal,Code,format,__VA_ARGS__) class LogicDevice; //class ccos_mqtt_connection; using ccos_mqtt_callback = std::function; //typedef int MQTT_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message); using mqtt_topic_list = std::list; using mqtt_msg_list = std::list; using ccos_mqtt_connection = std::tuple< void*, // MQTT_CLT_ID (mqtt_client_t*) mqtt_topic_list*, // MQTT_TIPIC_LIST_ID void*, // MSG_HOOK_ID (ccos_mqtt_msg_filter*) const char*, // CLINET_ID_ID ccos_mqtt_callback, // USER_MSG_CAKBCK_ID pthread_t, // CONNECTED_THREAD_ID (Linux使用pthread_t) CcosLock*, // CONN_SEND_LOCK_ID mqtt_msg_list*, // MSG_LIST_ID sem_t* // SEMAPHORE_HANDLE_ID (Linux使用sem_t*) >; const int MQTT_CLT_ID = 0; const int MQTT_TIPIC_LIST_ID = 1; const int MSG_HOOK_ID = 2; const int CLINET_ID_ID = 3; const int USER_MSG_CAKBCK_ID = 4; const int CONNECTED_HANDLE_ID = 5; const int CONN_SEND_LOCK_ID = 6; const int MSG_LIST_ID = 7; const int SEMAPHORE_HANDLE_ID = 8; using ccos_mqtt_msg_filter_func = std::function; using ccos_mqtt_msg_filter = std::tuple< ccos_mqtt_msg_filter_func, // FILTER_FUNC_ID std::shared_ptr, // LinuxEvent (替代 HANDLE) ResDataObject*, // FILTER_RES_OBJ_ID ResDataObject* // FILTER_RESPONS_OBJ_ID >; const int FILTER_FUNC_ID = 0; const int FILTER_HANDLE_ID = 1; const int FILTER_RES_OBJ_ID = 2; const int FILTER_RESPONS_OBJ_ID = 3; /// /// 逻辑设备命令收发接口 /// class LOGICDEVICE_API LogicDeviceSysIF { protected: LogicDevice *m_pLogicDev; public: LogicDeviceSysIF(void); virtual ~LogicDeviceSysIF(void); //init void SetLogicDevice(LogicDevice *p); LogicDevice* GetLogicDevice(); //Command In and Out //notify from lower layer /// /// notify from lower layer,设备上行命令,发送端为逻辑设备 /// /// /// virtual RET_STATUS HW_ACTION CmdFromLogicDev(ResDataObject PARAM_IN *pCmd); //notify to lower layer /// /// notify to lower layer,设备命令下行,接收端为逻辑设备 /// /// /// virtual RET_STATUS SYSTEM_CALL CmdToLogicDev(ResDataObject PARAM_IN *pCmd); }; using ccos_topic_filter = std::function; class LOGICDEVICE_API LogicDevice : public CcosLock, public Thread_Base { protected: char szPad2[128]; ccos_mqtt_connection* m_pMqttConntion; char szPad[128]; string m_strMqttUser; string m_strMqttPassword; string m_strServer; string m_strServerPort; bool m_bMqttUseSSL; ccos_topic_filter m_topicFilter; DWORD m_dwLastPacket; // int m_bAbstractOnline; //抽象设备是否在线,初始化启动是在线,可以通过Close下线 ResDataObject m_rsOnlineGroup; //在线设备组,在线状态,对应工作位 string m_strClientID; //MQTT 的客户端连接的ID,一般用DEVICE的逻辑ID来表示 VendorID_ProductID_SerialID_LogicID; std::vector m_strSubTopics; //要订阅的主题列表 bool m_bIsTreeRoot; string m_strEBusRoot; std::vector m_subDevices; //按照原有Ebus组织形式的树 string m_strCCOSDevicePath; //设备完整CCOS路径 CCOS/DEVICE/Detector/RF/ecomdemo/demo/1234 string m_strAbstractPath; //抽象设备路径:CCOS/DEVICE/Generator,可以被Open,可以执行Action和通知等各种CCOS消息 string m_strCCOSRoot; //设备根 ,CCOS/DEVICE ,CCOS/Driver , CCOS/Host,只可以被Open std::vector m_subCcosDevices; //按照新的树形结构的树 LogicDevice* m_pParent; //父节点 string m_strDevicePath; //相对CCOS root 根节点的 当前设备的名称,用于构建主题URI,必须以/开头 ResDataObject m_Actions; MsgQueue* m_pPacketReceivedQue; MsgQueue* m_pPacketSendingQue; //发送队列 string m_strCurTransaction; //Logger *m_pLogger; char *m_pDevInstance; ResDataObject *m_pResErrorList; std::shared_ptr m_EvtNotify;//此事件的目的是当本地有事件发生,而这事件需要要往上传递,需要驱动线程主动下来GET. sem_t m_SemphRequest; //请求到了 sem_t m_SemphPublish; //有包要发送 DriverDPC *m_pDrvDPC;//设备对象的HOST RET_STATUS SystemLog(SYSLOGLEVEL Level, const char *pCode, const char* fmt, ...); RET_STATUS IoSystemLog(int Level, const char* pCode, const char* pContext,size_t ContextSize, const char* pAppId = ""); bool CheckFeatureLicense(const char *pszFeatureId); virtual RET_STATUS HW_ACTION AddErrorMessageBase(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType = 0, const char* pAppId = ""); void SubscribeActions(); void SubscribeGroupActions(string wsName, int bOnline); void PacketArrived(ResDataObject* pRequest); static DWORD Thread_Publish_Thread(void* pPara); RET_STATUS InnerOpenClose(ResDataObject& req, ResDataObject& resp, bool openOrClose); public: LogicDevice(void); virtual ~LogicDevice(void); virtual bool OnStartThread() override; virtual bool OnEndThread() override; virtual bool Exec(void) override; RET_STATUS AddEbusChildren(LogicDevice* pChild, const char* szDevPath); RET_STATUS AddCcosChildren(LogicDevice* pChild, const char* szCcosDevPath); LogicDevice* GetEbusChild(const char* szEbusDevPath); LogicDevice* GetCcosChild(const char* szCcosDevPath); string& GetRootPath() { return m_strEBusRoot; } string& GetCcosRootPath() { return m_strCCOSDevicePath; } string& GetCcosAbstractPath() { return m_strAbstractPath; } void SetTreeRoot(bool bIsTreeRoot) { m_bIsTreeRoot = bIsTreeRoot; } bool IsTreeRoot() { return m_bIsTreeRoot; } void SetParent(LogicDevice* pParent) { m_pParent = pParent; } void SetEbusRoot(const char* pszEbusRoot) { m_strEBusRoot = pszEbusRoot; } bool HasAction(const char* pszAction) { return m_Actions.GetFirstOf(pszAction) >= 0; } bool GetActions(ResDataObject& resAction); void NotifyParent(ResDataObject* NotifyData, const char* pszTopic, ccos_mqtt_connection* hConnection=nullptr); void SetDrvDPC(DriverDPC *pDPC); DriverDPC *GetDrvDPC(); void SetClientRootID(const char* pszEBusRoot, const char* pszCCOSRoot); //设置 设备URI根 路径 ,包括EBUS和CCOS两个 virtual void OnSetClientID(); void SetTopicFilter(ccos_topic_filter filter); virtual void SubscribeSelf(); //所有子类必须使用 SubscribeTopic 订阅自身topic及资源topic,如果不知道需要DPC代订阅 void SubScribeTopic(const char* pszTopic, bool bSubscribe = true); void HelpSubscribeTopic(const char* pszTopic); //代注册接口,注册在主MQTT连接上 ccos_mqtt_connection* CreateConnection(const char* pszClientID, ccos_mqtt_callback onmsg); std::shared_ptr GetEvtHandle(); void NotifyDrvThread(); virtual RET_STATUS SYSTEM_CALL EvtProcedure();//m_EvtNotify事件发生后,由上层驱动线程进行调入. virtual int SYSTEM_CALL GetDevice_Thread_Priority();//return THREAD_PRIORITY_NONE //get device type,returns device type GUID virtual bool SYSTEM_CALL GetDeviceType(GUID &DevType) = 0; virtual void SYSTEM_CALL CompleteInit(); virtual void SYSTEM_CALL CompleteUnInit(); //get device resource /// /// 获取设备资源 get device resource /// /// /// virtual RET_STATUS SYSTEM_CALL GetDeviceResource(ResDataObject PARAM_OUT *pDeviceResource); //3. command //normal sync routine,Request to device and response from device /// /// 设备的命令响应接口 normal sync routine,Request to device and response from device /// /// /// /// virtual RET_STATUS SYSTEM_CALL Request(ResDataObject PARAM_IN *pRequest, ResDataObject PARAM_OUT *pResponse) = 0; //upper layer ------------------------------> lower layer //app -> service -> framework -> logicdevice -> hw //notify to lower layer /// /// 发送给设备的命令 只是通知吗?notify to lower layer /// /// /// virtual RET_STATUS SYSTEM_CALL CmdToLogicDev(ResDataObject PARAM_IN *pCmd) = 0; //处理来自客户的请求 virtual RET_STATUS ProcessSubscribeMsg(ResDataObject* pCmd); virtual RET_STATUS ProcessSubscribeRequest(ResDataObject* pCmd); virtual RET_STATUS ProcessRequest(ResDataObject* pCmd, PACKET_CMD cmd); virtual RET_STATUS ProcessResponse(ResDataObject* pCmd, PACKET_CMD cmd); //需要子类自己实现自己的订阅处理 virtual RET_STATUS ProcessNotify(ResDataObject* pCmd, PACKET_CMD cmd); //notify from lower layer /// /// 设备侧 来的 命令?只可能是通知吗?notify from lower layer /// /// /// RET_STATUS HW_ACTION CmdFromLogicDev(ResDataObject PARAM_IN *pCmd); virtual RET_STATUS HW_ACTION AddErrorMessageUnicode(const char* DevInstance, const char* Code, int &Level, const wchar_t* ResInfo, const wchar_t* Description, int nMessageType = 0); virtual RET_STATUS HW_ACTION AddErrorMessage(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType = 0, const char* pAppId = ""); virtual RET_STATUS HW_ACTION AddErrorMessage(const char* Code, int &Level, const char* ResInfo, int nMessageType = 0, const char* pAppId = ""); virtual RET_STATUS HW_ACTION DelErrorMessage(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType = 0); virtual RET_STATUS HW_ACTION DelErrorMessage(const char* Code, int &Level, const char* ResInfo, int nMessageType = 0); static ccos_mqtt_connection* NewConnection(const char* pszServer, const char* pszServerPort, const char* pszUserName, const char* pszPassword, const char* pszClientID, ccos_mqtt_callback onmsg); virtual RET_STATUS DevOpen(const char* pszDevUri, const char* pszGroup, ResDataObject& resRespons); virtual RET_STATUS DevClose(const char* pszDevUri, const char* pszGroup, ResDataObject& resRespons); virtual RET_STATUS DevGet(const char* pszDevUri, const char* pszProperty, ResDataObject& resRespons); virtual RET_STATUS DevSet(const char* pszDevUri, const char* pszProperty, const char* pszValueSet, ResDataObject& resRespons); virtual RET_STATUS DevUpdate(const char* pszDevUri, const char* pszProperty, const char* pszValueUpdate, ResDataObject& resRespons); virtual RET_STATUS DevAdd(const char* pszDevUri, const char* pszProperty, const char* pszValueAdd, ResDataObject& resRespons); virtual RET_STATUS DevDel(const char* pszDevUri, const char* pszProperty, const char* pszValueDel, ResDataObject& resRespons); virtual RET_STATUS DevAction(const char* pszDevUri, const char* pszActionName, const char* pszParams, ResDataObject& resRespons); virtual RET_STATUS DevMessage(const char* pszDevUri, const char* pszTopic, const char* pszMessageValue, ResDataObject& resRespons); }; //创建额外连接,需要提供回调函数 LOGICDEVICE_API ccos_mqtt_connection* NewConnection(const char* pszClientID, ccos_mqtt_callback onmsg, DWORD dwOpenTimeout = 10000, bool async = false); LOGICDEVICE_API void CloseConnection(ccos_mqtt_connection* hConnection); LOGICDEVICE_API void ResetConnection(ccos_mqtt_connection* hConnection); //主动订阅主题 LOGICDEVICE_API int SubscribeTopic(ccos_mqtt_connection* hConnection, const char* pszTopic, bool isShare = false); //主题订阅取消 LOGICDEVICE_API int UnSubscribe(ccos_mqtt_connection* hConnection, const char* pszTopic); //往指定主题发送CCOS协议包整包,使用临时创建连接,仅发送,不接收 LOGICDEVICE_API int PublishMsg(ResDataObject* pCmd, const char* pszTopic, const char* pszSenderName, DWORD dwTimeout = 3); //往指定主题发送CCOS协议包整包,使用已创建的连接发送 LOGICDEVICE_API int PublishAction(ResDataObject* pAction, const char* pszTopic, ccos_mqtt_connection* hConnection = nullptr, DWORD dwTimeout = 3); //使用现有连接 往指定主题发送Action包,携带参数,并指定答复的Topic,同步等待resp,超时没收到应答返回失败,复用链接时须小心,该函数会 //LOGICDEVICE_API int ActionAndRespWithConnection(ccos_mqtt_connection* hConnection, const char* pAction, ResDataObject& req, ResDataObject* pContext, const char* pszTopic, const char* pszRespTopic, ResDataObject& resObj, // DWORD dwWaitTime); //使用现有连接 往指定主题发送Action包,携带参数,复用连接订阅的默认Topic,通过提供的消息钩子 等待resp,超时没收到应答返回失败, LOGICDEVICE_API int ActionAndRespWithConnDefalt(ccos_mqtt_connection* hConnection, const char* pAction, ResDataObject& req, ResDataObject* pContext, const char* pszTopic, ResDataObject& resObj, DWORD dwWaitTime); //往指定主题发送Action包,携带参数,并指定答复的Topic,同步等待resp,临时创建连接,并等待应答,超时没收到应答返回失败 LOGICDEVICE_API int ActionAndResp(const char* pAction, ResDataObject* pContext, const char* pszTopic, const char* pszRespTopic, ResDataObject& resObj, DWORD dwWaitTime, const char* pszSenderName);