DriverPlatform-V3.1/LogicDevice--mqtt/LogicDevice.h

#pragma once

#define LOGICDEVICE_API 

#include <uuid/uuid.h>
#include "Definitions.h"
#include "ResDataObject.h"
//#include "logger.h"
#include "DPC.h"
#include "CcosLock.h"
#include <functional>
#include "CcosFileHandle.h"
#include "CcosThread.h"
#include "MsgQueue.h"
#include <list>

#include <sys/time.h> // 添加Linux时间头文件
#include <pthread.h>  // 添加Linux线程头文件
#include <unistd.h>   // 添加unistd头文件
#include <semaphore.h> // 添加信号量头文件

 
// 添加 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<void(ResDataObject*, const  char*, void*)>;

//typedef int MQTT_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message);
using mqtt_topic_list = std::list<string>;
using mqtt_msg_list = std::list<ResDataObject*>;
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*)
	std::string,           // 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*)
	std::atomic<bool>      //线程运行状态标志
>; 
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; 
const int THREAD_RUNNING_ID = 9;



using ccos_mqtt_msg_filter_func = std::function<bool(ResDataObject*)>;
using ccos_mqtt_msg_filter = std::tuple<
	ccos_mqtt_msg_filter_func,  // FILTER_FUNC_ID
	std::shared_ptr<LinuxEvent>,                // 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;

static std::vector<std::string> m_subscribedTopics;
/// <summary>
/// 逻辑设备命令收发接口
/// </summary>
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
	/// <summary>
	/// notify from lower layer，设备上行命令，发送端为逻辑设备
	/// </summary>
	/// <param name="pCmd"></param>
	/// <returns></returns>
	virtual RET_STATUS HW_ACTION CmdFromLogicDev(ResDataObject PARAM_IN *pCmd);

	//notify to lower layer
		/// <summary>
	/// notify to lower layer，设备命令下行，接收端为逻辑设备
	/// </summary>
	/// <param name="pCmd"></param>
	/// <returns></returns>
	virtual RET_STATUS SYSTEM_CALL CmdToLogicDev(ResDataObject PARAM_IN *pCmd);

};

using ccos_topic_filter = std::function<void(ResDataObject* pTopics, ResDataObject*pReq)>;

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<string> m_strSubTopics;		//要订阅的主题列表
	bool m_bIsTreeRoot;
	string m_strEBusRoot;

	std::vector<LogicDevice*> 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<LogicDevice*> m_subCcosDevices; //按照新的树形结构的树
	LogicDevice* m_pParent; //父节点
	string m_strDevicePath; //相对CCOS root 根节点的 当前设备的名称，用于构建主题URI，必须以/开头

	ResDataObject m_Actions;
	MsgQueue<ResDataObject>* m_pPacketReceivedQue;
	MsgQueue<ResDataObject>* m_pPacketSendingQue;  //发送队列

	string m_strCurTransaction;

	//Logger *m_pLogger;
	char *m_pDevInstance;
	ResDataObject *m_pResErrorList;
	std::shared_ptr<LinuxEvent> 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<LinuxEvent> 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
	/// <summary>
	/// 获取设备资源 get device resource
	/// </summary>
	/// <param name="pDeviceResource"></param>
	/// <returns></returns>
	virtual RET_STATUS SYSTEM_CALL GetDeviceResource(ResDataObject PARAM_OUT *pDeviceResource);

	//3. command
	//normal sync routine,Request to device and response from device

	/// <summary>
	/// 设备的命令响应接口 normal sync routine,Request to device and response from device
	/// </summary>
	/// <param name="pRequest"></param>
	/// <param name="pResponse"></param>
	/// <returns></returns>
	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
	/// <summary>
	/// 发送给设备的命令 只是通知吗？notify to lower layer
	/// </summary>
	/// <param name="pCmd"></param>
	/// <returns></returns>
	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
	/// <summary>
	/// 设备侧 来的 命令？只可能是通知吗？notify from lower layer
	/// </summary>
	/// <param name="pCmd"></param>
	/// <returns></returns>
	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);