From bae3a4fd9406635608edf0c0d16c52cf7ca06a66 Mon Sep 17 00:00:00 2001 From: fujuntang <fujuntang@smartai.com> Date: 星期六, 23 十月 2021 15:20:57 +0800 Subject: [PATCH] Optimize the code source. --- src/bh_api.cpp | 857 +++++++++++++++++++++++++++++++------------------------- 1 files changed, 476 insertions(+), 381 deletions(-) diff --git a/src/bh_api.cpp b/src/bh_api.cpp index 73c7772..37f8377 100644 --- a/src/bh_api.cpp +++ b/src/bh_api.cpp @@ -7,11 +7,12 @@ #include "bh_api.h" #include <pthread.h> #include <getopt.h> -#include "bhome_msg_api.pb.h" -#include "bhome_msg.pb.h" -#include "error_msg.pb.h" +#include "msg_mgr.h" +#include "../proto/source/error_msg.pb.h" #include "../proto/source/bhome_msg.pb.h" #include "../proto/source/bhome_msg_api.pb.h" + +#define TIME_WAIT 3 static Logger *logger = LoggerFactory::getLogger(); @@ -20,7 +21,42 @@ static pthread_mutex_t mutex; -static char errString[100] = { 0x00 }; +static pthread_t gTids; + +static void *client_run_check(void *skptr) { + pthread_detach(pthread_self()); + + int data; + int sec, nsec; + int rv; + int key; + char buf[MAX_STR_LEN] = { 0x00 }; + void *buf_temp = NULL; + int size; + + sec = TIME_WAIT; + nsec = 0; + sprintf(buf, "%s", "Success"); + data = net_mod_socket_int_get(gNetmod_socket); + while(true) { + + rv = net_mod_socket_recvfrom(gNetmod_socket, &buf_temp, &size, &key, SVR_STR, data); + if (rv == 0) { + + BHFree(buf_temp, size); + + rv = net_mod_socket_sendto_timeout(gNetmod_socket, buf, strlen(buf), key, sec, nsec, SVR_STR, data); + if (rv != 0) { + logger->error("the process check response failed with error: %s!\n", bus_strerror(rv)); + } + + } else { + + logger->error("the process check failed with error: %s!\n", bus_strerror(rv)); + + } + } +} int BHRegister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) { @@ -29,6 +65,7 @@ int count = 0; void *buf = NULL; int min = 0; + char *errString = NULL; ProcInfo pData; #if defined(PRO_DE_SERIALIZE) @@ -43,11 +80,9 @@ ::bhome_msg::ProcInfo input; if ((!input.ParseFromArray(proc_info, proc_info_len)) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + bus_errorset(rv); + + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -58,26 +93,27 @@ #else if ((proc_info == NULL) || (proc_info_len == 0) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x90, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + bus_errorset(rv); + + goto exit_entry; } #endif memset(&pData, 0x00, sizeof(ProcInfo)); if (gRun_stat == 0) { pthread_mutex_init(&mutex, NULL); + +#if defined(MSG_HANDLER) + msg_init(); +#endif } else { logger->error("the process has already registered!\n"); rv = EBUS_RES_BUSY; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } rv = pthread_mutex_trylock(&mutex); @@ -89,52 +125,62 @@ #if defined(PRO_DE_SERIALIZE) if (_input.proc_id != NULL) { count = strlen(_input.proc_id) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count; strncpy(pData.proc_id, _input.proc_id, min); } if (_input.name != NULL) { count = strlen(_input.name) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN -1) : count; strncpy(pData.name, _input.name, min); } if (_input.public_info != NULL) { count = strlen(_input.public_info) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1) : count; strncpy(pData.public_info, _input.public_info, min); } if (_input.private_info != NULL) { count = strlen(_input.private_info) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1): count; strncpy(pData.private_info, _input.private_info, min); } #else if (strlen((const char *)(((ProcInfo *)proc_info)->proc_id)) > 0) { count = strlen((const char *)(((ProcInfo *)proc_info)->proc_id)) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1): count; strncpy(pData.proc_id, ((ProcInfo *)proc_info)->proc_id, min); } if (strlen((const char *)(((ProcInfo *)proc_info)->name)) > 0) { count = strlen((const char *)(((ProcInfo *)proc_info)->name)) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count; strncpy(pData.name, ((ProcInfo *)proc_info)->name, min); } if (strlen((const char *)(((ProcInfo *)proc_info)->public_info)) > 0) { count = strlen((const char *)(((ProcInfo *)proc_info)->public_info)) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count; strncpy(pData.public_info, ((ProcInfo *)proc_info)->public_info, min); } if (strlen((const char *)(((ProcInfo *)proc_info)->private_info)) > 0) { count = strlen((const char *)(((ProcInfo *)proc_info)->private_info)) + 1; - min = count > MAX_STR_LEN ? MAX_STR_LEN : count; + min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count; strncpy(pData.private_info, ((ProcInfo *)proc_info)->private_info, min); } #endif + + if (pData.proc_id == NULL) { + rv = EBUS_INVALID_PARA; + + bus_errorset(rv); + + pthread_mutex_unlock(&mutex); + + goto exit_entry; + } gNetmod_socket = net_mod_socket_open(); hashtable_t *hashtable = mm_get_hashtable(); @@ -149,20 +195,20 @@ rv = net_mod_socket_reg(gNetmod_socket, &pData, sizeof(ProcInfo), NULL, 0, timeout_ms, PROC_REG); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); pthread_mutex_unlock(&mutex); } else { rv = EBUS_RES_BUSY; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } - + +exit_entry: + errString = bus_strerror(0, 1); #if defined(PRO_DE_SERIALIZE) ::bhome_msg::MsgCommonReply mcr; mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); @@ -180,9 +226,14 @@ *reply_len = min; #endif + + if (rv == 0) { + pthread_create(&gTids, NULL, client_run_check, NULL); + + return true; + } - return true; - + return false; } int BHUnregister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) @@ -190,6 +241,7 @@ int rv; int min; void *buf = NULL; + char *errString = NULL; #if defined(PRO_DE_SERIALIZE) struct _ProcInfo_proto @@ -205,11 +257,9 @@ if(!input.ParseFromArray(proc_info, proc_info_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; + bus_errorset(rv); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -219,11 +269,9 @@ #else if ((reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; + bus_errorset(rv); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + goto exit_entry; } #endif @@ -232,10 +280,9 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } rv = pthread_mutex_trylock(&mutex); @@ -250,21 +297,20 @@ gRun_stat = 0; } - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); pthread_mutex_unlock(&mutex); } else { rv = EBUS_RES_BUSY; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } +exit_entry: + errString = bus_strerror(0, 1); #if defined(PRO_DE_SERIALIZE) ::bhome_msg::MsgCommonReply mcr; mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); @@ -281,8 +327,12 @@ *reply = buf; *reply_len = min; #endif - - return true; + + if (rv == 0) { + return true; + } + + return false; } @@ -293,6 +343,7 @@ void *buf = NULL; int total = 0; int count = 0; + char *errString = NULL; char *topics_buf = NULL; #if defined(PRO_DE_SERIALIZE) @@ -306,11 +357,9 @@ if(!input.ParseFromArray(topics, topics_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input.amount = input.topic_list_size(); @@ -327,11 +376,9 @@ if ((topics == NULL) || (topics_len == 0) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } total = topics_len; @@ -341,22 +388,20 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } topics_buf = (char *)malloc(total); if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); logger->error("in BHRegisterTopics: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, total); @@ -382,10 +427,12 @@ rv = net_mod_socket_reg(gNetmod_socket, topics_buf, count, NULL, 0, timeout_ms, PROC_REG_TCS); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - free(topics_buf); + + bus_errorset(rv); + +exit_entry: + errString = bus_strerror(0, 1); #if defined(PRO_DE_SERIALIZE) ::bhome_msg::MsgCommonReply mcr; @@ -404,8 +451,11 @@ *reply_len = len; #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHQueryTopicAddress(const void *remote, const int remote_len, const void *topic, const int topic_len, void **reply, int *reply_len, const int timeout_ms) @@ -414,6 +464,7 @@ int min; void *buf = NULL; int size; + char *errString = NULL; char topics_buf[MAX_STR_LEN] = { 0x00 }; ProcInfo_query *ptr = NULL; ProcInfo *Proc_ptr = NULL; @@ -428,16 +479,25 @@ }_input0; const char *_input1; + + struct _MsgQueryTopicReply + { + std::string proc_id; + + unsigned long long mq_id; + long long abs_addr; + std::string ip; + int port; + } mtr_list[128]; + int mtr_list_num = 0; ::bhome_msg::BHAddress input0; ::bhome_msg::MsgQueryTopic input1; if (!input0.ParseFromArray(remote, remote_len) || !input1.ParseFromArray(topic, topic_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -449,11 +509,9 @@ #else if ((topic == NULL) || (topic_len == 0) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -461,10 +519,9 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -477,21 +534,8 @@ #endif rv = net_mod_socket_reg(gNetmod_socket, topics_buf, min, &buf, &size, timeout_ms, PROC_QUE_TCS); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - + bus_errorset(rv); #if defined(PRO_DE_SERIALIZE) - - struct _MsgQueryTopicReply - { - std::string proc_id; - - unsigned long long mq_id; - long long abs_addr; - std::string ip; - int port; - }mtr_list[128]; - int mtr_list_num = 0; if (rv == 0) { @@ -501,15 +545,19 @@ if (mtr_list_num > sizeof(mtr_list) / sizeof(mtr_list[0])) { mtr_list_num = sizeof(mtr_list) / sizeof(mtr_list[0]); } - + + Proc_ptr = &(ptr->procData); for(int i = 0; i < mtr_list_num; i++) { - mtr_list[i].proc_id = ptr->procData.proc_id; + mtr_list[i].proc_id = (Proc_ptr + i)->proc_id; mtr_list[i].mq_id = ID_RSV; mtr_list[i].abs_addr = ABS_ID_RSV; mtr_list[i].ip = "127.0.0.1"; mtr_list[i].port = 5000; } } + +exit_entry: + errString = bus_strerror(0, 1); ::bhome_msg::MsgQueryTopicReply mtr; mtr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); @@ -544,8 +592,11 @@ #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHQueryProcs(const void *remote, const int remote_len, const void *query, const int query_len, void **reply, int *reply_len, const int timeout_ms) @@ -554,6 +605,7 @@ void *buf = NULL; int size; int min; + char *errString = NULL; ProcInfo_sum *Proc_ptr = NULL; char data_buf[MAX_STR_LEN] = { 0x00 }; @@ -573,10 +625,9 @@ if (!input0.ParseFromArray(remote, remote_len) || !input1.ParseFromArray(query, query_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -587,10 +638,9 @@ #else if ((reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -598,10 +648,9 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + bus_errorset(rv); + + goto exit_entry; } if (query != NULL) { @@ -609,10 +658,9 @@ } rv = net_mod_socket_reg(gNetmod_socket, data_buf, strlen(data_buf), &buf, &size, timeout_ms, PROC_QUE_ATCS); - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - + bus_errorset(rv); + +exit_entry: #if defined(PRO_DE_SERIALIZE) struct _MsgQueryProcReply { @@ -657,29 +705,32 @@ } } } - - ::bhome_msg::MsgQueryProcReply mpr; - mpr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); - mpr.mutable_errmsg()->set_errstring(errString); - - for(int i = 0; i < mpr_list_num; i++) - { - ::bhome_msg::MsgQueryProcReply_Info *mpri = mpr.add_proc_list(); - mpri->mutable_proc()->set_proc_id(mpr_list[i].proc_id); - mpri->mutable_proc()->set_name(mpr_list[i].name); - mpri->mutable_proc()->set_public_info(mpr_list[i].public_info); - mpri->mutable_proc()->set_private_info(mpr_list[i].private_info); - mpri->set_online(mpr_list[i].online); - for(int j = 0; j < mpr_list[i].topic_list_num; j++) - { - mpri->mutable_topics()->add_topic_list(mpr_list[i].topic_list[j]); - } - } - - *reply_len = mpr.ByteSizeLong(); - *reply = malloc(*reply_len); - mpr.SerializePartialToArray(*reply, *reply_len); } + + errString = bus_strerror(0, 1); + + ::bhome_msg::MsgQueryProcReply mpr; + mpr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); + mpr.mutable_errmsg()->set_errstring(errString); + + for(int i = 0; i < mpr_list_num; i++) + { + ::bhome_msg::MsgQueryProcReply_Info *mpri = mpr.add_proc_list(); + mpri->mutable_proc()->set_proc_id(mpr_list[i].proc_id); + mpri->mutable_proc()->set_name(mpr_list[i].name); + mpri->mutable_proc()->set_public_info(mpr_list[i].public_info); + mpri->mutable_proc()->set_private_info(mpr_list[i].private_info); + mpri->set_online(mpr_list[i].online); + for(int j = 0; j < mpr_list[i].topic_list_num; j++) + { + mpri->mutable_topics()->add_topic_list(mpr_list[i].topic_list[j]); + } + } + + *reply_len = mpr.ByteSizeLong(); + *reply = malloc(*reply_len); + mpr.SerializePartialToArray(*reply, *reply_len); + #else if (rv == 0) { *reply = buf; @@ -695,8 +746,11 @@ } #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) @@ -707,6 +761,7 @@ int count = 0; int len, i; void *buf = NULL; + char *errString = NULL; char *topics_buf = NULL; #if defined(PRO_DE_SERIALIZE) @@ -720,10 +775,9 @@ if(!input.ParseFromArray(topics, topics_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input.amount = input.topic_list_size(); @@ -741,10 +795,9 @@ #else if ((topics == NULL) || (topics_len == 0) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -752,22 +805,20 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } topics_buf = (char *)malloc(total); if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); logger->error("in BHSubscribeTopics: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, total); @@ -806,8 +857,8 @@ } - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); +exit_entry: + errString = bus_strerror(0, 1); free(topics_buf); @@ -827,9 +878,12 @@ *reply = buf; *reply_len = len; #endif - - return true; + + if (rv == 0) { + return true; + } + return false; } int BHSubscribeNetTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) @@ -848,6 +902,7 @@ int BHHeartbeat(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) { int rv; + char *errString = NULL; #if defined(PRO_DE_SERIALIZE) struct _ProcInfo_proto @@ -862,11 +917,9 @@ if(!input.ParseFromArray(proc_info,proc_info_len)) { rv = EBUS_INVALID_PARA; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -875,8 +928,10 @@ _input.private_info = input.private_info().c_str(); rv = 0; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); + +exit_entry: + errString = bus_strerror(0, 1); ::bhome_msg::MsgCommonReply mcr; mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); @@ -886,7 +941,11 @@ mcr.SerializePartialToArray(*reply,*reply_len); #endif - return true; + if (rv == 0) { + return true; + } + + return false; } #if defined(PRO_DE_SERIALIZE) @@ -912,8 +971,7 @@ if(!input.ParseFromArray(msgpub, msgpub_len)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); return false; } @@ -924,10 +982,9 @@ if ((topic == NULL) || (content == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - - return false; + bus_errorset(rv); + + return false; } #endif @@ -935,8 +992,7 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); return false; } @@ -965,10 +1021,7 @@ if (rv > 0) return true; - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - + return false; } @@ -977,7 +1030,7 @@ int rv; int len; void *buf; - int key; + int key = 0; int size; int sec, nsec; char buf_temp[100] = { 0x00 }; @@ -995,18 +1048,16 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } if ((msgpub == NULL) || (msgpub_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } if (timeout_ms > 0) { @@ -1035,12 +1086,11 @@ if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHReadSub: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, len + 10); @@ -1052,14 +1102,13 @@ if (data_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHReadSub: Out of memory!\n"); free(topics_buf); - return false; + goto exit_entry; } memset(data_buf, 0x00, size - len + 10); @@ -1077,56 +1126,61 @@ rsr.data = topics_buf; } - sprintf(buf_temp, "%d", key); - - if ((proc_id != NULL) && (proc_id_len != NULL)) { - rsr.proc_id = buf_temp; - *proc_id_len = rsr.proc_id.size(); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, rsr.proc_id.data(), *proc_id_len); - } - free(topics_buf); free(data_buf); - - ::bhome_msg::MsgPublish Mp; - Mp.set_topic(rsr.topic); - Mp.set_data(rsr.data.data()); - *msgpub_len = Mp.ByteSizeLong(); - *msgpub = malloc(*msgpub_len); - Mp.SerializePartialToArray(*msgpub, *msgpub_len); -#else - void *ptr; - if (len < size) { - ptr = malloc(size - len); - len = size - len; - memcpy(ptr, data_buf, len); - } else { - ptr = malloc(len); - memcpy(ptr, topics_buf, len); - } - *msgpub = ptr; - *msgpub_len = len; - - free(topics_buf); - free(data_buf); - - if ((proc_id != NULL) && (proc_id_len != NULL)) { - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", key); - - *proc_id_len = strlen(buf_temp); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, buf_temp, *proc_id_len); - } - -#endif - - } else { - - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); } + +exit_entry: + sprintf(buf_temp, "%d", key); + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + rsr.proc_id = buf_temp; + *proc_id_len = rsr.proc_id.size(); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, rsr.proc_id.data(), *proc_id_len); + } + + if (rv != 0) { + rsr.topic = STR_RSV; + if (data_buf != NULL) { + rsr.data = data_buf; + } else { + rsr.data = STR_RSV; + } + } + + ::bhome_msg::MsgPublish Mp; + Mp.set_topic(rsr.topic); + Mp.set_data(rsr.data.data()); + *msgpub_len = Mp.ByteSizeLong(); + *msgpub = malloc(*msgpub_len); + Mp.SerializePartialToArray(*msgpub, *msgpub_len); +#else + void *ptr; + if (len < size) { + ptr = malloc(size - len); + len = size - len; + memcpy(ptr, data_buf, len); + } else { + ptr = malloc(len); + memcpy(ptr, topics_buf, len); + } + *msgpub = ptr; + *msgpub_len = len; + + free(topics_buf); + free(data_buf); + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", key); + + *proc_id_len = strlen(buf_temp); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, buf_temp, *proc_id_len); + } + +#endif if (rv == 0) return true; @@ -1137,15 +1191,17 @@ int BHAsyncRequest(const void *remote, const int remote_len, const void *request, const int request_len, void **msg_id, int *msg_id_len) { int rv; - void *buf; + void *buf = NULL; int size; - int val; + int val = 0; int len; int min; int data; int sec, nsec; + std::string str; std::string MsgID; int timeout_ms = 3000; + char data_buf[MAX_STR_LEN] = { 0x00 }; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; @@ -1169,10 +1225,9 @@ if (!input0.ParseFromArray(remote, remote_len) || !input1.ParseFromArray(request, request_len)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -1186,10 +1241,9 @@ if ((request == NULL) || (request_len == 0)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -1197,10 +1251,9 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -1208,14 +1261,30 @@ #else strncpy(buf_temp, (char *)request, (sizeof(buf_temp) - 1) > strlen((char *)request) ? strlen((char *)request) : (sizeof(buf_temp) - 1)); #endif - - rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + + str = buf_temp; + val = net_mod_socket_buf_data_get(gNetmod_socket, str); + if(val > 0) { + + rv = 0; + + } else { + rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + if (rv == 0) { + + len = size > (sizeof(data_buf) - 1) ? (sizeof(data_buf) - 1) : size; + memcpy(data_buf, (char *)buf, len); + val = atoi((char *)data_buf); + if (val > 0) { + str = buf_temp; + net_mod_socket_buf_data_set(gNetmod_socket, str, val); + } + + free(buf); + } + } + if (rv == 0) { - - val = atoi((char *)buf); - - free(buf); - if (val > 0) { len = strlen(buf_temp) + 1; @@ -1227,12 +1296,11 @@ if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHAsyncRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, len); @@ -1267,9 +1335,9 @@ } } - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); +exit_entry: if((msg_id == NULL) || (msg_id_len == NULL)) { if (rv == 0) return true; @@ -1277,18 +1345,16 @@ return false; } - if (rv == 0) { - - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", val); - MsgID = buf_temp; + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", val); + MsgID = buf_temp; - *msg_id_len = MsgID.size(); - *msg_id = malloc(*msg_id_len); - memcpy(*msg_id, MsgID.data(), *msg_id_len); + *msg_id_len = MsgID.size(); + *msg_id = malloc(*msg_id_len); + memcpy(*msg_id, MsgID.data(), *msg_id_len); + if (rv == 0) return true; - } return false; @@ -1298,17 +1364,20 @@ void **reply, int *reply_len, const int timeout_ms) { int rv; - void *buf; + void *buf = NULL; int size; int val; int min, len; int data; + std::string str; net_node_t node; int node_size; int recv_arr_size; + char data_buf[MAX_STR_LEN] = { 0x00 }; net_mod_recv_msg_t *recv_arr; net_mod_err_t *errarr; int errarr_size = 0; + char *errString = NULL; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; @@ -1338,27 +1407,26 @@ if (!input0.ParseFromArray(remote, remote_len) || !input1.ParseFromArray(request, request_len) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } + memset(&node, 0x00, sizeof(node)); _input0.mq_id = input0.mq_id(); - _input0.abs_addr = input0.abs_addr(); - _input0.ip = input0.ip().c_str(); - _input0.port = input0.port(); - _input1.topic = input1.topic().c_str(); - _input1.data = input1.data().c_str(); + _input0.abs_addr = input0.abs_addr(); + _input0.ip = input0.ip().c_str(); + _input0.port = input0.port(); + _input1.topic = input1.topic().c_str(); + _input1.data = input1.data().c_str(); #else if ((request == NULL) || (request_len == 0) || (reply == NULL) || (reply_len == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -1366,10 +1434,9 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -1377,43 +1444,57 @@ #else strncpy(buf_temp, (char *)request, (sizeof(buf_temp) - 1) > request_len ? request_len : (sizeof(buf_temp) - 1)); #endif - - rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + + str = buf_temp; + val = net_mod_socket_buf_data_get(gNetmod_socket, str); + if(val > 0) { + + rv = 0; + + } else { + rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + if (rv == 0) { + + len = size > (sizeof(data_buf) - 1) ? (sizeof(data_buf) - 1) : size; + memcpy(data_buf, (char *)buf, len); + val = atoi((char *)data_buf); + if (val > 0) { + str = buf_temp; + net_mod_socket_buf_data_set(gNetmod_socket, str, val); + } + + free(buf); + } + } + if (rv == 0) { - val = atoi((char *)buf); - - free(buf); - if (val > 0) { memset(&node, 0x00, sizeof(node)); - + len = strlen(buf_temp) + 1; #if defined(PRO_DE_SERIALIZE) len += strlen(_input1.data); #endif - data = net_mod_socket_svr_get(gNetmod_socket); topics_buf = (char *)malloc(len); if (topics_buf == NULL) { - + rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - + bus_errorset(rv); + logger->error("in BHRequest: Out of memory!\n"); - - return false; + + goto exit_entry; } memset(topics_buf, 0x00, len); - + strncpy(topics_buf, buf_temp, strlen(buf_temp) + 1); #if defined(PRO_DE_SERIALIZE) strncpy(topics_buf + strlen(buf_temp) + 1, _input1.data, strlen(_input1.data)); -#endif +#endif node.key = val; - if (timeout_ms > 0) { rv = net_mod_socket_sendandrecv_timeout(gNetmod_socket, &node, 1, topics_buf, len, &recv_arr, &recv_arr_size, &errarr, &errarr_size, timeout_ms); @@ -1434,14 +1515,15 @@ size = recv_arr[0].content_length; buf = (char *)malloc(size); if (buf == NULL) { - printf("Out of memory\n"); + logger->error("in BHRequest: Out of memory!\n"); - exit(0); + free(topics_buf); + + goto exit_entry; } memset((char *)buf, 0x00, size); strncpy((char *)buf, (char *)recv_arr[0].content, size); - } net_mod_socket_free_recv_msg_arr(recv_arr, recv_arr_size); @@ -1451,42 +1533,40 @@ } rv = 0; - } else { + rv = EBUS_TIMEOUT; } } else { rv = EBUS_RES_UNSUPPORT; } - + free(topics_buf); } + + bus_errorset(rv); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + if ((proc_id != NULL) && (proc_id_len != NULL)) { + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", node.key); + + rr.proc_id = buf_temp; + *proc_id_len = rr.proc_id.size(); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, rr.proc_id.c_str(), *proc_id_len); + } if (rv == 0) { - if ((proc_id != NULL) && (proc_id_len != NULL)) { - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", node.key); - - rr.proc_id = buf_temp; - *proc_id_len = rr.proc_id.size(); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, rr.proc_id.c_str(), *proc_id_len); - } - topics_buf = (char *)malloc(size + 10); if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); logger->error("in BHRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, size + 10); @@ -1495,31 +1575,33 @@ free(buf); free(topics_buf); + } else { + + rr.data = STR_RSV; } +exit_entry: + errString = bus_strerror(0, 1); + #if defined(PRO_DE_SERIALIZE) - if (rv == 0) { - ::bhome_msg::MsgRequestTopicReply mrt; - mrt.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); - mrt.mutable_errmsg()->set_errstring(errString); - mrt.set_data(rr.data.data()); - *reply_len = mrt.ByteSizeLong(); - *reply = malloc(*reply_len); - mrt.SerializePartialToArray(*reply, *reply_len); - } + ::bhome_msg::MsgRequestTopicReply mrt; + mrt.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); + mrt.mutable_errmsg()->set_errstring(errString); + mrt.set_data(rr.data.data()); + *reply_len = mrt.ByteSizeLong(); + *reply = malloc(*reply_len); + mrt.SerializePartialToArray(*reply, *reply_len); #else - if (rv == 0) { - min = strlen(errString) + 1; - buf = malloc(min); - memcpy(buf, errString, strlen(errString)); - *((char *)buf + min - 1) = '\0'; + min = strlen(errString) + 1; + buf = malloc(min); + memcpy(buf, errString, strlen(errString)); + *((char *)buf + min - 1) = '\0'; - *reply = buf; - *reply_len = min; - } + *reply = buf; + *reply_len = min; #endif - if (rv == 0) + if (rv == 0) return true; return false; @@ -1536,23 +1618,29 @@ int sec, nsec; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; - + + struct _ReadRequestReply + { + std::string proc_id; + std::string topic; + std::string data; + void *src; + } rrr; + if (gRun_stat == 0) { logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } if ((request == NULL) || (request_len == 0) || (src == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); - return false; + goto exit_entry; } data = net_mod_socket_svr_get(gNetmod_socket); @@ -1573,14 +1661,7 @@ } if (rv == 0) { - struct _ReadRequestReply - { - std::string proc_id; - std::string topic; - std::string data; - void *src; - } rrr; - + if ((proc_id != NULL) && (proc_id_len != NULL)) { sprintf(buf_temp, "%d", key); rrr.proc_id = buf_temp; @@ -1594,12 +1675,11 @@ if (topics_buf == NULL) { rv = EBUS_NO_MEM; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); logger->error("in BHReadRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, size + MIN_STR_LEN); @@ -1617,29 +1697,34 @@ rrr.data = topics_buf + len + 1; free(topics_buf); - -#if defined(PRO_DE_SERIALIZE) - ::bhome_msg::MsgRequestTopic mrt; - mrt.set_topic(rrr.topic); - mrt.set_data(rrr.data.data()); - *request_len = mrt.ByteSizeLong(); - *request = malloc(*request_len); - mrt.SerializePartialToArray(*request,*request_len); -#else - *request = buf; - *request_len = size; -#endif - free(buf); - buf = malloc(sizeof(int)); - *(int *)buf = key; - *src = buf; + bus_errorset(rv); } - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); - +exit_entry: + + if (rv != 0) { + rrr.topic = STR_RSV; + rrr.data = STR_RSV; + } + +#if defined(PRO_DE_SERIALIZE) + ::bhome_msg::MsgRequestTopic mrt; + mrt.set_topic(rrr.topic); + mrt.set_data(rrr.data.data()); + *request_len = mrt.ByteSizeLong(); + *request = malloc(*request_len); + mrt.SerializePartialToArray(*request,*request_len); +#else + *request = buf; + *request_len = size; +#endif + + buf = malloc(sizeof(int)); + *(int *)buf = key; + *src = buf; + if (rv == 0) return true; @@ -1650,6 +1735,8 @@ { int rv; int data; + int sec = 3; + int nsec = 0; const char *_input; #if defined(PRO_DE_SERIALIZE) @@ -1657,8 +1744,7 @@ if (!input.ParseFromArray(reply, reply_len) || (src == NULL)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); return false; } @@ -1669,8 +1755,7 @@ if ((src == NULL) || (reply == NULL) || (reply_len == 0)) { rv = EBUS_INVALID_PARA; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); return false; } @@ -1683,17 +1768,15 @@ logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); return false; } data = net_mod_socket_svr_get(gNetmod_socket); - rv = net_mod_socket_sendto(gNetmod_socket, _input, strlen(_input), *(int *)src, SVR_STR, data); + rv = net_mod_socket_sendto_timeout(gNetmod_socket, _input, strlen(_input), *(int *)src, sec, nsec, SVR_STR, data); - memset(errString, 0x00, sizeof(errString)); - strncpy(errString, bus_strerror(rv), sizeof(errString)); + bus_errorset(rv); if (rv == 0) return true; @@ -1713,6 +1796,7 @@ int BHGetLastError(void **msg, int *msg_len) { void *buf = NULL; + char *errString = bus_strerror(0, 1); buf = malloc(strlen(errString) + 1); @@ -1729,3 +1813,14 @@ return false; } + +#if defined(MSG_HANDLER) +int inter_key_get(void) +{ + if (gNetmod_socket != NULL) + return net_mod_socket_get_key(gNetmod_socket); + + return 0; +} +#endif + -- Gitblit v1.8.0