修改帧结构

applications/ems_datahubs/datahubs.vcxproj

@@ -84,6 +84,7 @@
   </PropertyGroup>
   <ItemGroup>
     <ClCompile Include="iconv-utils.cpp" />
+    <ClCompile Include="kutilities.cpp" />
     <ClCompile Include="main.cpp" />
     <ClCompile Include="openjson.cpp" />
     <ClCompile Include="opmysql.cpp" />
@@ -92,6 +93,7 @@
     <ClInclude Include="frame_define.h" />
     <ClInclude Include="iconv-utils.h" />
     <ClInclude Include="kdefine.h" />
+    <ClInclude Include="kutilities.h" />
     <ClInclude Include="mqtt_msg.h" />
     <ClInclude Include="openjson.h" />
     <ClInclude Include="opmysql.h" />

applications/ems_datahubs/frame_define.h

@@ -5,17 +5,43 @@
 
 NAMESPACE_BEGIN(HJ)
 
-typedef enum tagFrameType : char
+typedef enum tagErrorCode : unsigned char
 {
-	Frame_Request = 0,  //请求帧
+	ERR_OK = 0X00,
-	Frame_Response = 1, //返回帧
+	ERR_INVALID_LEN = 0X01,
+	ERR_INVALID_UTF8 = 0X02,
+	ERR_INVALID_BUF_LEN = 0X03,
+	ERR_UNKOWN = 0XFF
+}ErrorCode;
+
+typedef enum tagFrameType : unsigned char
+{
+	Frame_Request = 0x00,  //请求帧
+	Frame_Response = 0x01, //返回帧
+	Frame_DeviceData_Request = 0x02, //来自采集程序的数据请求包，将数据保存到数据库中
 }FrameType;
 
+typedef struct tagFrameTail
+{
+	unsigned char frame_delimiter[4] = { 0xEE,0xFF,0xEE,0xFF };
+}FrameTail;
+
 typedef struct tagFrame
 {
 	FrameType frame_type;					//帧类型
 	unsigned int frame_len;					//帧数据长度
-	char frame_content[1];					//帧的内容，实际应为json字符串，由json内容自解释
+	unsigned char frame_content[1];					//帧的内容，实际应为json字符串，由json内容自解释
+	tagFrame()
+	{
+		frame_type = Frame_Response;
+		frame_len = 1;
+		frame_content[0] = ERR_OK;
+	}
+	void setErrorFrame(ErrorCode err = ERR_OK)
+	{
+		frame_len = 1;
+		frame_content[0] = err;
+	}
 }MessageFrame;
 
 NAMESPACE_END(HJ)

applications/ems_datahubs/kdefine.h

@@ -36,6 +36,6 @@
 #define NAMESPACE_BEGIN(X) namespace X {
 #define NAMESPACE_END(X) }
 #define USING_NAMESPACE(X)  using namespace X
-
+#define USING_NAMESPACE_HJ	using namespace HJ
 
 #endif

applications/ems_datahubs/main.cpp

@@ -1,11 +1,16 @@
 #include <string>
-#include <chrono>
 #include <fcntl.h> 
 #include <unistd.h> 
 #include <fstream>
-#include <memory>
 #include <iomanip>
-#include <sstream>
+
+#include "kdefine.h"
+#include "frame_define.h"
+#include "kutilities.h"
+#include "mqtt_msg.h"
+#include "opmysql.h"
+#include "iconv-utils.h"
+#include "openjson.h"
 
 #include <hv/hv.h>
 #include <hv/hmain.h>
@@ -13,176 +18,24 @@
 #include <hv/hloop.h>
 #include <hv/hsocket.h>
 #include <hv/hssl.h>
-#include <zlib/zlib.h>
-
-#include "kdefine.h"
-
-#include "mqtt_msg.h"
-#include "openjson.h"
-#include "opmysql.h"
-
-#include "iconv-utils.h"
 
+#ifndef TEST_UNPACK
 #define TEST_UNPACK 1
-
+#endif
-
-#define CHUNK 16384
-
-/* Compress from file source to file dest until EOF on source.
-def() returns Z_OK on success, Z_MEM_ERROR if memory could not be
-allocated for processing, Z_STREAM_ERROR if an invalid compression
-level is supplied, Z_VERSION_ERROR if the version of zlib.h and the
-version of the library linked do not match, or Z_ERRNO if there is
-an error reading or writing the files. */
-int CompressString(const char* in_str, size_t in_len, std::string& out_str, int level)
-{
-    if (!in_str)
-        return Z_DATA_ERROR;
-
-    int ret, flush;
-    unsigned have;
-    z_stream strm;
-
-    unsigned char out[CHUNK];
-
-    /* allocate deflate state */
-    strm.zalloc = Z_NULL;
-    strm.zfree = Z_NULL;
-    strm.opaque = Z_NULL;
-    ret = deflateInit(&strm, level);
-    if (ret != Z_OK)
-        return ret;
-
-    std::shared_ptr<z_stream> sp_strm(&strm, [](z_stream* strm)
-        {
-            (void)deflateEnd(strm);
-        });
-
-    const char* end = in_str + in_len;
-
-    //size_t pos_index = 0;
-    size_t distance = 0;
-    /* compress until end of file */
-    do
-    {
-        distance = end - in_str;
-        strm.avail_in = (distance >= CHUNK) ? CHUNK : distance;
-        strm.next_in = (Bytef*)in_str;
-
-        // next pos
-        in_str += strm.avail_in;
-        flush = (in_str == end) ? Z_FINISH : Z_NO_FLUSH;
-
-        /* run deflate() on input until output buffer not full, finish
-        compression if all of source has been read in */
-        do
-        {
-            strm.avail_out = CHUNK;
-            strm.next_out = out;
-            ret = deflate(&strm, flush);  /* no bad return value */
-            if (ret == Z_STREAM_ERROR)
-                break;
-            have = CHUNK - strm.avail_out;
-            out_str.append((const char*)out, have);
-        } while (strm.avail_out == 0);
-
-        if (strm.avail_in != 0);   /* all input will be used */
-        break;
-
-        /* done when last data in file processed */
-    } while (flush != Z_FINISH);
-
-    if (ret != Z_STREAM_END) /* stream will be complete */
-        return Z_STREAM_ERROR;
-
-    /* clean up and return */
-    return Z_OK;
-}
-
-/* Decompress from file source to file dest until stream ends or EOF.
-inf() returns Z_OK on success, Z_MEM_ERROR if memory could not be
-allocated for processing, Z_DATA_ERROR if the deflate data is
-invalid or incomplete, Z_VERSION_ERROR if the version of zlib.h and
-the version of the library linked do not match, or Z_ERRNO if there
-is an error reading or writing the files. */
-int DecompressString(const char* in_str, size_t in_len, std::string& out_str)
-{
-    if (!in_str)
-        return Z_DATA_ERROR;
-
-    int ret;
-    unsigned have;
-    z_stream strm;
-    unsigned char out[CHUNK];
-
-    /* allocate inflate state */
-    strm.zalloc = Z_NULL;
-    strm.zfree = Z_NULL;
-    strm.opaque = Z_NULL;
-    strm.avail_in = 0;
-    strm.next_in = Z_NULL;
-    ret = inflateInit(&strm);
-    if (ret != Z_OK)
-        return ret;
-
-    std::shared_ptr<z_stream> sp_strm(&strm, [](z_stream* strm)
-        {
-            (void)inflateEnd(strm);
-        });
-
-    const char* end = in_str + in_len;
-
-    //size_t pos_index = 0;
-    size_t distance = 0;
-
-    int flush = 0;
-    /* decompress until deflate stream ends or end of file */
-    do
-    {
-        distance = end - in_str;
-        strm.avail_in = (distance >= CHUNK) ? CHUNK : distance;
-        strm.next_in = (Bytef*)in_str;
-
-        // next pos
-        in_str += strm.avail_in;
-        flush = (in_str == end) ? Z_FINISH : Z_NO_FLUSH;
-
-        /* run inflate() on input until output buffer not full */
-        do
-        {
-            strm.avail_out = CHUNK;
-            strm.next_out = out;
-            ret = inflate(&strm, Z_NO_FLUSH);
-
-            if (ret == Z_STREAM_ERROR) /* state not clobbered */
-                break;
-
-            switch (ret)
-            {
-            case Z_NEED_DICT:
-                ret = Z_DATA_ERROR;   /* and fall through */
-            case Z_DATA_ERROR:
-            case Z_MEM_ERROR:
-                return ret;
-            }
-            have = CHUNK - strm.avail_out;
-            out_str.append((const char*)out, have);
-        } while (strm.avail_out == 0);
-
-        /* done when inflate() says it's done */
-    } while (flush != Z_FINISH);
-
-    /* clean up and return */
-    return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
-}
-
-
-//static OpDatabase gOpDatabase;
 
 #if TEST_UNPACK
 static unpack_setting_t unpack_setting;
 #endif
 
+//连接关闭回调
+static void on_close(hio_t* io);
+//接入回调
+static void on_accept(hio_t* io);
+//接收到数据
+static void on_recv(hio_t* io, void* buf, int readbytes);
+//重载参数回调
+static void on_reload(void* userdata);
+
 /*
  * @build: make
  * @usage: datahubs -h
@@ -208,6 +61,7 @@ typedef struct conf_ctx_s
     std::string dbuser;
     std::string dbname;
 } conf_ctx_t;
+
 conf_ctx_t g_conf_ctx;
 
 inline void conf_ctx_init(conf_ctx_t* ctx)
@@ -421,12 +275,6 @@ int parse_confile(const char* confile)
     return 0;
 }
 
-static void on_reload(void* userdata)
-{
-    hlogi("reload confile [%s]", g_main_ctx.confile);
-    parse_confile(g_main_ctx.confile);
-}
-
 //////1///////////////////////////////
 #if 0
 #define LOCKFILE "/var/lock/datahub.lock"
@@ -593,165 +441,124 @@ int main(int argc, char** argv)
     return 0;
 }
 
+void worker_fn(void* userdata)
+{
+    conf_ctx_t* ptrCtx = (conf_ctx_t*)(intptr_t)(userdata);
+    long port = ptrCtx->port;
+
+    //initialize database connection
+    bool dbok = OpDatabase::getInstance()->OpenDatabase(ptrCtx->dbserver,ptrCtx->dbuser,ptrCtx->dbname);
+    if (!dbok)
+    {
+        hloge("failed to open database, exit now...");
+        return;
+    }
+
+    hlogi("database connection created!");
+
+    hloop_t* loop = hloop_new(0);
+    const char* host = ptrCtx->host.c_str();
+    hio_t* listenio = hloop_create_tcp_server(loop, host, port, on_accept);
+
+    if (listenio == NULL)
+    {
+        hloge("worker process finished");
+        return;
+    }
+
+    hlogi("port=%ld pid=%ld tid=%ld listenfd=%d", port, hv_getpid(), hv_gettid(), hio_fd(listenio));
+
+    hloop_run(loop);
+
+    hlogi("database connection close!");
+    OpDatabase::getInstance()->CloseDatabase();
+
+    hloop_free(&loop);
+}
+
+
+/// 各回调函数定义
+
+static void on_reload(void* userdata)
+{
+    hlogi("reload confile [%s]", g_main_ctx.confile);
+    parse_confile(g_main_ctx.confile);
+}
+
 static void on_close(hio_t* io)
 {
     hlogi("on_close fd=%d error=%d", hio_fd(io), hio_error(io));
 }
 
-std::string get_current_timestamp()
+//接入回调
+static void on_accept(hio_t* io)
 {
-    auto now = std::chrono::system_clock::now();
+    hlogi("on_accept connfd=%d", hio_fd(io));
-    //通过不同精度获取相差的毫秒数
-    uint64_t dis_millseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count()
-        - std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count() * 1000;
-    time_t tt = std::chrono::system_clock::to_time_t(now);
-    auto time_tm = localtime(&tt);
-    char strTime[25] = { 0 };
-    sprintf(strTime, "%d-%02d-%02d %02d:%02d:%02d.%03d", time_tm->tm_year + 1900,
-        time_tm->tm_mon + 1, time_tm->tm_mday, time_tm->tm_hour,
-        time_tm->tm_min, time_tm->tm_sec, (int)dis_millseconds);
-    return std::string(strTime);
-}
 
-// 函数用于将内存块转换为十六进制字符串
+    char localaddrstr[SOCKADDR_STRLEN] = { 0 };
-std::string printHex(const void* data, size_t size)
+    char peeraddrstr[SOCKADDR_STRLEN] = { 0 };
-{
+    hlogi("accept connfd=%d [%s] <=== [%s]", hio_fd(io),
-    std::ostringstream oss;
+        SOCKADDR_STR(hio_localaddr(io), localaddrstr),
-    std::ostringstream oss2;
+        SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
-    std::ostringstream ossrow;
 
-    const size_t lineSize = 16; // 每行输出的字节数
+    hio_setcb_close(io, on_close);
-    const unsigned char* p = static_cast<const unsigned char*>(data);
+    hio_setcb_read(io, on_recv);
 
-    int ic = 0;
+#if TEST_UNPACK
-    int row = 0;
+    hio_set_unpack(io, &unpack_setting);
-
-    ossrow << std::setw(8) << std::setfill('0') << std::hex << row++ << "h : ";
-    oss << ossrow.str().c_str();
-
-    for (size_t i = 0; i < size; ++i)
-    {
-        ic++;
-
-        // 每个字节之间用空格分隔
-        oss << std::setw(2) << std::setfill('0') << std::hex << std::uppercase << static_cast<int>(p[i]);
-
-        char ch = (isprint(p[i]) != 0) ? p[i] : '.';
-
-        oss2 << ch;
-
-        // 每lineSize个字节换行
-        if ((i + 1) % lineSize == 0)
-        {
-            ossrow.clear();
-            ossrow.str("");
-
-            oss << "    [" << oss2.str().c_str() << "]" << std::endl;
-            oss2.clear();
-            oss2.str("");
-
-            ossrow << std::setw(8) << std::setfill('0') << std::hex << row++ << "h : ";
-            oss << ossrow.str().c_str();
-
-            ic = 0;
-        }
-        else if (i == size - 1)
-        {
-            if ((i + 1) % lineSize != 0)
-            {
-                if (i % 2 != 0)
-                {
-                    for (size_t j = 0; j < (lineSize - ic); j++)
-                    {
-                        oss << " --";
-                    }
-                }
-                else
-                {
-                    for (size_t j = 0; j < (lineSize - ic); j++)
-                    {
-                        oss << " --";
-                    }
-                }
-            }
-            oss << "    [" << oss2.str().c_str();
-
-            if ((i + 1) % lineSize != 0)
-            {
-                for (size_t j = 0; j < (lineSize - ic); j++)
-                {
-                    oss << " ";
-                }
-            }
-
-            oss << "]" << std::endl;
-            oss2.clear();
-            oss2.str("");
-
-            ic = 0;
-        }
-#if 0
-        else if ((i + 1) % 8 == 0)
-        {
-            oss << "    ";
-            oss2 << "  ";
-        }
 #endif
-        else
-        {
-            oss << " ";
-        }
-    }
-    return oss.str();
-}
 
+    hio_read_start(io);
+}
 
 //接收到数据
 static void on_recv(hio_t* io, void* buf, int readbytes)
 {
+    USING_NAMESPACE_HJ;
+
     char localaddrstr[SOCKADDR_STRLEN] = { 0 };
     char peeraddrstr[SOCKADDR_STRLEN] = { 0 };
     hlogi("### 1 ### on_recv fd=%d readbytes=%d [%s] <==== [%s]", hio_fd(io), readbytes,
         SOCKADDR_STR(hio_localaddr(io), localaddrstr),
         SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
 
-    char ret[1] = { 0 };
+    MessageFrame respFrame;
-    if (readbytes > 0xFFFF - 1)
+    if( readbytes > 0xFFFF - 1 )
     {
         hloge("too large data buffer to process: %d", readbytes);
-        ret[0] = 1;
+        respFrame.setErrorFrame(ERR_INVALID_BUF_LEN);
-        hio_write(io, (void*)ret, 1);
+        hio_write(io, (void*)&respFrame, sizeof respFrame);
         return;
     }
+    
+    MessageFrame *pReadFrame = (MessageFrame*)buf;
 
-    MessageData* pData = (MessageData*)buf;
+    if( pReadFrame->frame_len > 0xFFFF - 1 )
+    {
+        hloge("too big string buffer to process: %d, it should be less than %d", pReadFrame->frame_len, 0xFFFF);
+        respFrame.setErrorFrame(ERR_INVALID_LEN);
+        hio_write(io, (void*)&respFrame, sizeof respFrame);
+        return;
+    }
 
     OpenJson json;
 
-    if (pData->content_len > 0xFFFF - 1)
+    CODING buf_code = GetCoding((unsigned char*)pReadFrame->frame_content, pReadFrame->frame_len);   //判断是否是utf-8
-    {
-        hloge("too big string buffer to process: %d, it should be less than %d", pData->content_len, 0xFFFF);
-        ret[0] = 1;
-        hio_write(io, (void*)ret, 1);
-        return;
-    }
-
-    CODING buf_code = GetCoding((unsigned char*)pData->content_data, pData->content_len);   //判断是否是utf-8
 
     hlogi("<=== recieve buffer code is [%d]", buf_code);
 
-    std::string msg(pData->content_data, pData->content_len);
+    std::string msg((char*)pReadFrame->frame_content, pReadFrame->frame_len);
 
-    if (buf_code == CODING::GBK
+    if( buf_code == CODING::GBK
-        || buf_code == CODING::UNKOWN)
+        || buf_code == CODING::UNKOWN )
     {
         std::string str_result;
         //转换为UTF8
-        if (!GBKToUTF8(msg, str_result))
+        if( !GBKToUTF8(msg, str_result) )
         {
             hloge("Failed to transfer code from GBK to UTF-8");
-            ret[0] = 1;
+            respFrame.setErrorFrame(ERR_INVALID_UTF8);
-            hio_write(io, (void*)ret, 1);
+            hio_write(io, (void*)&respFrame, sizeof respFrame);
             return;
         }
 
@@ -798,22 +605,21 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
 
     //hlogd("<=== decode OK [\n%s\n]", printHex(pTmp, len).c_str());
 
-    if (!json.decode(msg))
+    if( !json.decode(msg) )
     {
         //delete[] pTmp;
         hloge("Failed to decode json string pack , length=%d", readbytes);
-        ret[0] = 1;
+        hio_write(io, (void*)&respFrame, sizeof respFrame);
-        hio_write(io, (void*)ret, 1);
         return;
     }
 
-    hio_write(io, (void*)ret, 1);
+    hio_write(io, (void*)&respFrame, sizeof respFrame);
 
     std::string fsucode = json["FsuCode"].s();
     std::string msg_type = json["type"].s();
     std::string timestamp = get_current_timestamp(); // json["TimeStamp"].s();
 
-    if (fsucode.length() == 0)
+    if( fsucode.length() == 0 )
     {
         //delete[] pTmp;
         hlogw("!!empty fsucode recieved!");
@@ -829,7 +635,7 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
     std::string out_compress;
 
     int zip_ret = 0;
-    if ((zip_ret = CompressString(pTmp, len, out_compress, Z_DEFAULT_COMPRESSION)) != Z_OK)
+    if( (zip_ret = CompressString(pTmp, len, out_compress, Z_DEFAULT_COMPRESSION)) != Z_OK )
     {
         hloge("Failed to compress source data, zip return value %d", zip_ret);
         return;
@@ -842,15 +648,15 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
 #endif
     //std::string msg2(pTmp, len);
 
-    if (msg_type == "gateway-data"
+    if( msg_type == "gateway-data"
         || msg_type == "gateway-alarmdata"
         || msg_type == "gateway-writedata"
         || msg_type == "gateway-readdata"
         || msg_type == "web-write"
-        || msg_type == "web-alarm")
+        || msg_type == "web-alarm" )
     {
         auto& IdCodeContent = json["IdCodeContent"];
-        if (IdCodeContent.size() <= 0)
+        if( IdCodeContent.size() <= 0 )
         {
             //delete[] pTmp;
             hloge("invalid IdCodeContent's size: %d", IdCodeContent.size());
@@ -860,13 +666,13 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
         auto& pNode = IdCodeContent[0]; //这是只解析第一个节点
         std::string oid = pNode["OID"].s();
 
-        OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
+        //OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
     }
 
-    if (msg_type == "web-read")
+    if( msg_type == "web-read" )
     {
         auto& IdCodeContent = json["IdCodes"];
-        if (IdCodeContent.size() <= 0)
+        if( IdCodeContent.size() <= 0 )
         {
             hloge("invalid IdCodes's size: %d", IdCodeContent.size());
             //delete[] pTmp;
@@ -876,63 +682,8 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
         auto& pNode = IdCodeContent[0]; //这是只解析第一个节点
         std::string oid = pNode.s();
 
-        OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
+        //OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
     }
     //delete[] pTmp;
 }
 
-//接入回调
-static void on_accept(hio_t* io)
-{
-    hlogi("on_accept connfd=%d", hio_fd(io));
-
-    char localaddrstr[SOCKADDR_STRLEN] = { 0 };
-    char peeraddrstr[SOCKADDR_STRLEN] = { 0 };
-    hlogi("accept connfd=%d [%s] <=== [%s]", hio_fd(io),
-        SOCKADDR_STR(hio_localaddr(io), localaddrstr),
-        SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
-
-    hio_setcb_close(io, on_close);
-    hio_setcb_read(io, on_recv);
-
-#if TEST_UNPACK
-    hio_set_unpack(io, &unpack_setting);
-#endif
-
-    hio_read_start(io);
-}
-
-void worker_fn(void* userdata)
-{
-    conf_ctx_t* ptrCtx = (conf_ctx_t*)(intptr_t)(userdata);
-    long port = ptrCtx->port;
-
-    //initialize database connection
-    bool dbok = OpDatabase::getInstance()->OpenDatabase(ptrCtx->dbserver,ptrCtx->dbuser,ptrCtx->dbname);
-    if (!dbok)
-    {
-        hloge("failed to open database, exit now...");
-        return;
-    }
-
-    hlogi("database connection created!");
-
-    hloop_t* loop = hloop_new(0);
-    const char* host = ptrCtx->host.c_str();
-    hio_t* listenio = hloop_create_tcp_server(loop, host, port, on_accept);
-
-    if (listenio == NULL)
-    {
-        hloge("worker process finished");
-        return;
-    }
-
-    hlogi("port=%ld pid=%ld tid=%ld listenfd=%d", port, hv_getpid(), hv_gettid(), hio_fd(listenio));
-
-    hloop_run(loop);
-
-    hlogi("database connection close!");
-    OpDatabase::getInstance()->CloseDatabase();
-
-    hloop_free(&loop);
-}

applications/ems_datahubs/openjson.cpp

@@ -188,10 +188,10 @@ bool OpenJson::Box::remove(OpenJson* node)
 //JsonContext
 OpenJson::Context::Context()
     :
-    offset_(0),
     data_(0),
-    root_(0),
+    size_(0),
-    size_(0)
+    offset_(0),
+    root_(0)
 {
 }
 
@@ -278,10 +278,10 @@ OpenJson::OpenJson(JsonType type)
     :type_(type),
      context_(0),
      wcontext_(0),
-     box_(0),
      idx_(0),
-     key_(0),
      len_(0),
+     box_(0),
+     key_(0),
      segment_(0)
 {
 }