#ifdef HAVE_CONFIG_H # include #endif #include "aqhome/msg/msg_ping.h" #include "aqhome/msg/endpointmgr.h" #include "aqhome/msg/endpoint_node.h" #include "aqhome/msg/endpoint_log.h" #include "aqhome/msg/endpoint_tty.h" #include "aqhome/ipc/endpoint_node_ipc_tcp.h" #include "aqhome/mqtt/endpoint_mqttc.h" #include "aqhome/mqtt/msg_mqtt_connect.h" #include "aqhome/mqtt/msg_mqtt_connack.h" #include "aqhome/mqtt/msg_mqtt_publish.h" #include "aqhome/mqtt/msg_mqtt_pubresponse.h" #include "aqhome/msgmanager.h" #include "aqhome/aqhome.h" #include #include #include #include #include #include #include #include GWEN_MSG *createPingMsg(uint8_t destAddr, uint8_t srcAddr) { GWEN_MSG *msg; int rv; msg=GWEN_Msg_new(AQH_MAXMSGSIZE); rv=GWEN_Msg_AddByte(msg, destAddr); if (rv<0) { fprintf(stderr, "ERROR1: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, 6); /* msglen */ if (rv<0) { fprintf(stderr, "ERROR2: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, AQH_MSG_TYPE_PING); /* ping */ if (rv<0) { fprintf(stderr, "ERROR3: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, srcAddr); /* src addr */ if (rv<0) { fprintf(stderr, "ERROR4: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */ if (rv<0) { fprintf(stderr, "ERROR5: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */ if (rv<0) { fprintf(stderr, "ERROR6: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */ if (rv<0) { fprintf(stderr, "ERROR7: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */ if (rv<0) { fprintf(stderr, "ERROR8: %d\n", rv); GWEN_Msg_free(msg); return NULL; } rv=AQH_NodeMsg_AddChecksum(msg); if (rv<0) { fprintf(stderr, "ERROR9: %d\n", rv); GWEN_Msg_free(msg); return NULL; } return msg; } int testEndpoints() { int rv; GWEN_MSG_ENDPOINT_MGR *emgr; GWEN_MSG_ENDPOINT *epTty; GWEN_MSG_ENDPOINT *epLog; GWEN_MSG_ENDPOINT *epTcp; rv=AQH_Init(); if (rv<0) { } emgr=AQH_MsgManager_new(0xc0); epTty=AQH_TtyNodeEndpoint_new("/dev/ttyUSB0", AQH_MSGMGR_ENDPOINTGROUP_NODE); if (epTty==NULL) { DBG_ERROR(NULL, "Error creating endpoint TTY"); return 2; } GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTty); epLog=AQH_LogEndpoint_new("endpoints.log", AQH_MSGMGR_ENDPOINTGROUP_NODE); if (epLog==NULL) { DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint LOG"); return 2; } GWEN_MsgEndpointMgr_AddEndpoint(emgr, epLog); epTcp=AQH_TcpdIpcNodeEndpoint_new("127.0.0.1", 45454, NULL, AQH_MSGMGR_ENDPOINTGROUP_IPC); if (epTcp==NULL) { DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP"); return 2; } GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTcp); for (;;) { DBG_DEBUG(AQH_LOGDOMAIN, "Next loop"); AQH_MsgManager_LoopOnce(emgr); } return 0; } int testMsgMqttConnect() { GWEN_MSG *msg; GWEN_BUFFER *dbuf; msg=GWEN_ConnectMqttMsg_new("MQTT", 4, AQH_MQTTMSG_CONNECT_FLAGS_USERNAME | AQH_MQTTMSG_CONNECT_FLAGS_PASSWD, 512, "CLIENTID123", "USER123", "PASSWD123"); if (msg==NULL) { DBG_ERROR(NULL, "Error creating message"); return 2; } dbuf=GWEN_Buffer_new(0, 256, 0, 1); AQH_ConnectMqttMsg_DumpToBuffer(msg, dbuf, "created"); fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(dbuf)); GWEN_Buffer_free(dbuf); GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2); return 0; } int testMqttConnection() { int rv; GWEN_MSG_ENDPOINT_MGR *emgr; GWEN_MSG_ENDPOINT *epTcp; GWEN_MSG *msgOut; rv=AQH_Init(); if (rv<0) { } //emgr=AQH_MsgManager_new(0xc0); emgr=GWEN_MsgEndpointMgr_new(); epTcp=AQH_MqttClientEndpoint_new("127.0.0.1", 1883, "MQTTClient", 0); if (epTcp==NULL) { DBG_ERROR(NULL, "Error creating endpoint TCPc"); return 2; } GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTcp); rv=GWEN_TcpcEndpoint_StartConnect(epTcp); if (rv<0) { DBG_ERROR(NULL, "Error starting connect (%d)", rv); return 2; } fprintf(stdout, "Sending CONNECT\n"); msgOut=GWEN_ConnectMqttMsg_new("MQTT", 4, 0, 10, "CLIENTID123", NULL, NULL); if (msgOut==NULL) { DBG_ERROR(NULL, "Error creating message"); return 2; } GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut); fprintf(stdout, "Waiting for response\n"); for (;;) { GWEN_MSG *msg; DBG_DEBUG(AQH_LOGDOMAIN, "Next loop"); //GWEN_MsgManager_LoopOnce(emgr); GWEN_MsgEndpointMgr_IoLoopOnce(emgr); msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp); if (msg) { if (AQH_MqttMsg_GetMsgTypeAndFlags(msg)==AQH_MQTTMSG_MSGTYPE_CONNACK) { GWEN_BUFFER *buf; buf=GWEN_Buffer_new(0, 256, 0, 1); AQH_ConnAckMqttMsg_DumpToBuffer(msg, buf, "received"); fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(buf)); GWEN_Buffer_free(buf); } else { DBG_ERROR(NULL, "Received this message:"); GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2); } GWEN_Msg_free(msg); break; } } fprintf(stdout, "Sending PUBLISH\n"); //msgOut=GWEN_PublishMqttMsg_new(AQH_MQTTMSG_FLAGS_QOS1, 1, "test/subject1", (const uint8_t*) "29.9", 4); msgOut=GWEN_PublishMqttMsg_new(0, 0, "test/subject1", (const uint8_t*) "29.9", 4); if (msgOut==NULL) { DBG_ERROR(NULL, "Error creating message"); return 2; } GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut); fprintf(stdout, "Waiting for response\n"); for (;;) { GWEN_MSG *msg; DBG_DEBUG(AQH_LOGDOMAIN, "Next loop"); //GWEN_MsgManager_LoopOnce(emgr); GWEN_MsgEndpointMgr_IoLoopOnce(emgr); msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp); if (msg) { if (AQH_MqttMsg_GetMsgTypeAndFlags(msg)==AQH_MQTTMSG_MSGTYPE_PUBACK) { GWEN_BUFFER *buf; buf=GWEN_Buffer_new(0, 256, 0, 1); AQH_PubResponseMqttMsg_DumpToBuffer(msg, buf, "received"); fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(buf)); GWEN_Buffer_free(buf); } else { DBG_ERROR(NULL, "Received this message:"); GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2); } GWEN_Msg_free(msg); break; } } return 0; } int main(int argc, char **argv) { //return testEndpoints(); //return testMsgMqttConnect(); return testMqttConnection(); }