From 2561a007b8d8999a4750046d0cfb3b1ad5af50ac Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期二, 09 四月 2024 15:29:32 +0800
Subject: [PATCH] test for perf

---
 src/bh_api.cpp |  740 ++++++++++++++++++++++++++++++++++----------------------
 1 files changed, 452 insertions(+), 288 deletions(-)

diff --git a/src/bh_api.cpp b/src/bh_api.cpp
index 312b203..c450bc2 100644
--- a/src/bh_api.cpp
+++ b/src/bh_api.cpp
@@ -3,13 +3,13 @@
 #include "bus_server_socket_wrapper.h"
 #include "shm_mm_wrapper.h"
 #include "proc_def.h"
+#include "mm.h"
 #include "usg_common.h"
 #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"
 
@@ -18,14 +18,15 @@
 static Logger *logger = LoggerFactory::getLogger();
 
 static int gRun_stat = 0;
+static int gRun_flag = true;
 static void *gNetmod_socket = NULL;
 
 static pthread_mutex_t mutex;
 
 static pthread_t gTids;
 
-static void *client_run_check(void *skptr) {
-  
+
+static void *client_run_check(void *skptr) { 
   pthread_detach(pthread_self());
  
   int data;
@@ -38,19 +39,24 @@
 
   sec = TIME_WAIT;
   nsec = 0;
-  sprintf(buf, "%s", "Success");
+  sprintf(buf, "%s", STR_EXEC);
   data = net_mod_socket_int_get(gNetmod_socket);
-  while(true) {
+  while(gRun_flag == true) {
     
     rv = net_mod_socket_recvfrom(gNetmod_socket, &buf_temp, &size, &key, SVR_STR, data);
     if (rv == 0) {
       
-      BHFree(buf_temp, size);
+      if (strncmp((char *)buf_temp, STR_RET, strlen(STR_RET)) != 0) {
 
-      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));
+        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 {
+        gRun_flag = false;
       }
+
+      BHFree(buf_temp, size);
       
     } else {
       
@@ -84,7 +90,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
 	_input.proc_id = input.proc_id().c_str();
@@ -97,13 +103,17 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    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");
@@ -111,7 +121,7 @@
     rv = EBUS_RES_BUSY;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
 
   rv = pthread_mutex_trylock(&mutex);
@@ -121,25 +131,25 @@
     shm_mm_wrapper_init(SHM_RES_SIZE);
     
 #if defined(PRO_DE_SERIALIZE)
-    if (_input.proc_id != NULL) {
+    if (strlen(_input.proc_id) > 0) {
       count = strlen(_input.proc_id) + 1;
       min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count;
       strncpy(pData.proc_id, _input.proc_id, min);
     }
 
-    if (_input.name != NULL) {
+    if (strlen(_input.name) > 0) {
       count = strlen(_input.name) + 1;
       min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN -1) : count;
       strncpy(pData.name, _input.name, min); 
     }
 
-    if (_input.public_info != NULL) {
+    if (strlen(_input.public_info) > 0) {
       count = strlen(_input.public_info) + 1;
       min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1) : count;
       strncpy(pData.public_info, _input.public_info, min);
     }
  
-    if (_input.private_info != NULL) {
+    if (strlen(_input.private_info) > 0) {
       count = strlen(_input.private_info) + 1;
       min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1): count;
       strncpy(pData.private_info, _input.private_info, min);
@@ -170,26 +180,27 @@
     }
 #endif 
 
-    if (pData.proc_id == NULL) {
+    if (strlen(pData.proc_id) == 0) {
       rv = EBUS_INVALID_PARA;
 
       bus_errorset(rv);
 
+      gRun_stat = 0;
       pthread_mutex_unlock(&mutex);
       
-      return false;
+      goto exit_entry;
     }
 
     gNetmod_socket = net_mod_socket_open();
     hashtable_t *hashtable = mm_get_hashtable();
     key = hashtable_alloc_key(hashtable);
+    net_mod_socket_bind(gNetmod_socket, key);
     count = hashtable_alloc_key(hashtable);
     rv = hashtable_alloc_key(hashtable);
     net_mod_socket_int_set(gNetmod_socket, count);
     net_mod_socket_svr_set(gNetmod_socket, rv);
     sprintf(pData.int_info, "%d", count);
     sprintf(pData.svr_info, "%d", rv);
-    net_mod_socket_bind(gNetmod_socket, key);
   
     rv = net_mod_socket_reg(gNetmod_socket, &pData, sizeof(ProcInfo), NULL, 0, timeout_ms, PROC_REG);
 
@@ -202,11 +213,11 @@
     rv = EBUS_RES_BUSY;
     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));
@@ -225,18 +236,25 @@
     
 #endif 
   
-  pthread_create(&gTids, NULL, client_run_check, NULL);
+  if (rv == 0) {
+    gRun_flag = true;
+    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)
 {
   int rv;
   int min;
+  int data;
+  int diff;
   void *buf = NULL;
   char *errString = NULL;
+  struct timeval start, end;
 
 #if defined(PRO_DE_SERIALIZE)
   struct _ProcInfo_proto
@@ -254,7 +272,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
   
 	_input.proc_id = input.proc_id().c_str();
@@ -266,7 +284,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
 #endif 
   
@@ -277,14 +295,30 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
 
   rv = pthread_mutex_trylock(&mutex);
   if (rv == 0) {
     rv = net_mod_socket_reg(gNetmod_socket, NULL, 0, NULL, 0, timeout_ms, PROC_UNREG);
     if (rv == 0) {
-   
+      gettimeofday(&start, NULL);
+      data = net_mod_socket_int_get(gNetmod_socket);
+      rv = net_mod_socket_sendto_timeout(gNetmod_socket, STR_RET, strlen(STR_RET), data, 3, 0);
+      if (rv != 0) { 
+        logger->error("the process check response failed with error: %s!\n", bus_strerror(rv));
+      }
+
+      while(gRun_flag == true) {
+        sleep(1);
+
+        gettimeofday(&end, NULL);
+
+        diff = end.tv_sec - start.tv_sec;
+        if (diff >= TIME_DUR)
+          break;
+      };
+
       net_mod_socket_close(gNetmod_socket);
       
       gNetmod_socket = NULL;
@@ -301,9 +335,10 @@
     rv = EBUS_RES_BUSY;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
+exit_entry:
   errString = bus_strerror(0, 1);
 #if defined(PRO_DE_SERIALIZE)
   ::bhome_msg::MsgCommonReply mcr;
@@ -321,8 +356,12 @@
   *reply = buf;
   *reply_len = min;
 #endif 
-  
-  return true;
+ 
+  if (rv == 0) {
+    return true;
+  }
+
+  return false;
   
 }
 
@@ -349,7 +388,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
 
 	_input.amount = input.topic_list_size();
@@ -368,7 +407,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
   total = topics_len;
@@ -380,7 +419,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
 
   topics_buf = (char *)malloc(total);
@@ -391,7 +430,7 @@
     
     logger->error("in BHRegisterTopics: Out of memory!\n");
 
-    return false;
+    goto exit_entry;
   }
   memset(topics_buf, 0x00, total);
 
@@ -420,6 +459,8 @@
   free(topics_buf);
   
   bus_errorset(rv);
+
+exit_entry:
   errString = bus_strerror(0, 1);
   
 #if defined(PRO_DE_SERIALIZE)
@@ -439,8 +480,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)
@@ -464,6 +508,17 @@
 	}_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;
@@ -471,7 +526,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
   _input0.mq_id = input0.mq_id();
@@ -485,7 +540,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
 #endif 
   
@@ -495,7 +550,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
 
 #if defined(PRO_DE_SERIALIZE)
@@ -511,35 +566,33 @@
   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) {
     
-    ptr = (ProcInfo_query *)((char *)buf + sizeof(int));
-    mtr_list_num = ptr->num;
+    min = *(int *)buf;
+    if (min > 0) {
+      ptr = (ProcInfo_query *)((char *)buf + sizeof(int));
+      mtr_list_num = ptr->num;
+      
+      if (mtr_list_num > sizeof(mtr_list) / sizeof(mtr_list[0])) {
+        mtr_list_num = sizeof(mtr_list) / sizeof(mtr_list[0]);
+      }
     
-    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 = (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;
+      }
+    } else {
+      mtr_list_num = 0;
     }
-    
-    for(int i = 0; i < mtr_list_num; i++) {
-      mtr_list[i].proc_id = ptr->procData.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;
-    }
+
+    free(buf);
   }
   
+exit_entry:
   errString = bus_strerror(0, 1);
   
   ::bhome_msg::MsgQueryTopicReply mtr;
@@ -575,8 +628,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)
@@ -607,7 +663,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
 	_input0.mq_id = input0.mq_id();
@@ -620,7 +676,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
 #endif 
   
@@ -630,7 +686,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
 
   if (query != NULL) {
@@ -639,7 +695,8 @@
   
   rv = net_mod_socket_reg(gNetmod_socket, data_buf, strlen(data_buf), &buf, &size, timeout_ms, PROC_QUE_ATCS);
   bus_errorset(rv);
-  
+
+exit_entry:
 #if defined(PRO_DE_SERIALIZE)
   struct _MsgQueryProcReply
 	{
@@ -663,52 +720,57 @@
     if (mpr_list_num > (sizeof(mpr_list) / sizeof(mpr_list[0]))) {
       mpr_list_num = sizeof(mpr_list) / sizeof(mpr_list[0]);
     }
-    
-    Proc_ptr = (ProcInfo_sum *)((char *)buf + sizeof(int));
-    for(int i = 0; i < mpr_list_num; i++) {
-      mpr_list[i].proc_id = (Proc_ptr + i)->procData.proc_id;
-      mpr_list[i].name = (Proc_ptr + i)->procData.name;
-      mpr_list[i].public_info = (Proc_ptr + i)->procData.public_info;
-      mpr_list[i].private_info = (Proc_ptr + i)->procData.private_info;
-      mpr_list[i].online = (Proc_ptr + i)->stat;
-      mpr_list[i].topic_list_num = (Proc_ptr + i)->list_num;
-      
-      for(int j = 0; j < mpr_list[i].topic_list_num; j++)
-      {
-        if (j == 0) {
-          mpr_list[i].topic_list[j] = (Proc_ptr + i)->reg_info;
-        } else if (j == 1) {
-          mpr_list[i].topic_list[j] = (Proc_ptr + i)->local_info;
-        } else if (j == 2) {
-          mpr_list[i].topic_list[j] = (Proc_ptr + i)->net_info;
+   
+    if (mpr_list_num > 0) {
+      Proc_ptr = (ProcInfo_sum *)((char *)buf + sizeof(int));
+      for(int i = 0; i < mpr_list_num; i++) {
+        mpr_list[i].proc_id = (Proc_ptr + i)->procData.proc_id;
+        mpr_list[i].name = (Proc_ptr + i)->procData.name;
+        mpr_list[i].public_info = (Proc_ptr + i)->procData.public_info;
+        mpr_list[i].private_info = (Proc_ptr + i)->procData.private_info;
+        mpr_list[i].online = (Proc_ptr + i)->stat;
+        mpr_list[i].topic_list_num = (Proc_ptr + i)->list_num;
+        
+        for(int j = 0; j < mpr_list[i].topic_list_num; j++)
+        {
+          if (j == 0) {
+            mpr_list[i].topic_list[j] = (Proc_ptr + i)->reg_info;
+          } else if (j == 1) {
+            mpr_list[i].topic_list[j] = (Proc_ptr + i)->local_info;
+          } else if (j == 2) {
+            mpr_list[i].topic_list[j] = (Proc_ptr + i)->net_info;
+          }
         }
       }
     }
-    
-    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);
+    free(buf);
   }
+    
+  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;
@@ -724,8 +786,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)
@@ -752,7 +817,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
   
   _input.amount = input.topic_list_size();
@@ -772,7 +837,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-    return false;
+    goto exit_entry;
   }
 #endif 
 
@@ -782,7 +847,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;    
   }
   
   topics_buf = (char *)malloc(total);
@@ -793,7 +858,7 @@
     
     logger->error("in BHSubscribeTopics: Out of memory!\n");
 
-    return false;
+    goto exit_entry;
   }
   memset(topics_buf, 0x00, total);
  
@@ -832,6 +897,7 @@
   
   }
  
+exit_entry:
   errString = bus_strerror(0, 1);
  
   free(topics_buf);
@@ -840,8 +906,8 @@
   ::bhome_msg::MsgCommonReply mcr;
 	mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv));
 	mcr.mutable_errmsg()->set_errstring(errString);
-	*reply_len=mcr.ByteSizeLong();
-	*reply=malloc(*reply_len);
+	*reply_len = mcr.ByteSizeLong();
+	*reply = malloc(*reply_len);
 	mcr.SerializePartialToArray(*reply,*reply_len);
 #else 
   len = strlen(errString) + 1;
@@ -852,9 +918,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)
@@ -890,7 +959,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-		return false;
+		goto exit_entry;
   }
   
 	_input.proc_id = input.proc_id().c_str();
@@ -900,6 +969,8 @@
 
   rv = 0;
   bus_errorset(rv);
+
+exit_entry:
   errString = bus_strerror(0, 1);
   
   ::bhome_msg::MsgCommonReply mcr;
@@ -910,7 +981,11 @@
 	mcr.SerializePartialToArray(*reply,*reply_len);
 #endif 
 
-  return true;
+  if (rv == 0) {
+    return true;
+  }
+
+  return false;
 }
 
 #if defined(PRO_DE_SERIALIZE)
@@ -948,8 +1023,8 @@
     
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
-    
-		return false;
+  
+    return false;
   }
 #endif 
   
@@ -986,9 +1061,7 @@
 
   if (rv > 0)
     return true;
-  
-  bus_errorset(rv);
-  
+ 
   return false;
 }
 
@@ -997,7 +1070,7 @@
   int rv;
   int len;
   void *buf;
-  int key;
+  int key = 0;
   int size;
   int sec, nsec;
   char buf_temp[100] = { 0x00 };
@@ -1017,14 +1090,14 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
   
   if ((msgpub == NULL) || (msgpub_len == NULL)) {
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-		return false;
+    goto exit_entry;
   }
 
   if (timeout_ms > 0) {
@@ -1055,9 +1128,9 @@
       rv = EBUS_NO_MEM;
       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);
     
@@ -1071,11 +1144,11 @@
         rv = EBUS_NO_MEM;
         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);
 
@@ -1093,56 +1166,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 {
-
-    bus_errorset(rv);
-
   }
+  
+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;
@@ -1153,18 +1231,21 @@
 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;
-  
+  hashtable_t *hashtable = mm_get_hashtable();
+
 #if defined(PRO_DE_SERIALIZE)
   struct _BHAddress
 	{
@@ -1187,7 +1268,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
   
-		return false;
+    goto exit_entry;
   }
 	
 	_input0.mq_id = input0.mq_id();
@@ -1203,7 +1284,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
   
-		return false;
+    goto exit_entry;
   }
 #endif 
   
@@ -1213,7 +1294,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
  
 #if defined(PRO_DE_SERIALIZE)
@@ -1221,14 +1302,35 @@
 #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) && (hashtable_get(hashtable, val) != NULL)) {
+
+    rv = 0;
+
+  } else {
+
+    if ((val > 0) && (hashtable_get(hashtable, val) == NULL)) {
+      net_mod_socket_buf_data_del(gNetmod_socket, str); 
+    }
+
+    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;
@@ -1242,9 +1344,9 @@
         rv = EBUS_NO_MEM;
         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);
       
@@ -1280,6 +1382,8 @@
   }
 
   bus_errorset(rv);
+
+exit_entry:
   if((msg_id == NULL) || (msg_id_len == NULL)) { 
     if (rv == 0)
       return true;
@@ -1287,18 +1391,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;
 
@@ -1308,21 +1410,24 @@
               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;
-  
+  hashtable_t *hashtable = mm_get_hashtable();
+
   struct _RequestReply
   {
     std::string proc_id;
@@ -1351,9 +1456,10 @@
     rv = EBUS_INVALID_PARA;
     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();
@@ -1367,7 +1473,7 @@
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-		return false;
+    goto exit_entry;
   }
 #endif 
   
@@ -1377,7 +1483,7 @@
     rv = EBUS_RES_NO;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
   
 #if defined(PRO_DE_SERIALIZE)
@@ -1385,42 +1491,59 @@
 #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) && (hashtable_get(hashtable, val) != NULL)) {
+    rv = 0;
+  } else {
+    if ((val > 0) && (hashtable_get(hashtable, val) == NULL)) {
+      net_mod_socket_buf_data_del(gNetmod_socket, str);
+    }
+
+    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;
         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);
@@ -1441,14 +1564,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);
@@ -1458,30 +1582,31 @@
         }
 
         rv = 0;
-
       } else {
+
         rv = EBUS_TIMEOUT; 
       }
       
     } else {
       rv = EBUS_RES_UNSUPPORT; 
     }
-
+   
     free(topics_buf);
   }
-
+  
   bus_errorset(rv);
+
+  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) {
       
@@ -1490,7 +1615,7 @@
       
       logger->error("in BHRequest: Out of memory!\n");
 
-      return false;
+      goto exit_entry;
     }
     memset(topics_buf, 0x00, size + 10);
     
@@ -1499,33 +1624,40 @@
     
     free(buf);
     free(topics_buf);
+  } else {
+
+    rr.data = STR_RSV;
   }
 
+exit_entry:
   errString = bus_strerror(0, 1);
   
+  if (rv != 0) {
+    if ((proc_id != NULL) && (proc_id_len != NULL)) {
+      *proc_id_len = 0;
+      *proc_id = NULL;
+    }
+  }
+  
 #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; 
@@ -1542,21 +1674,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;
     bus_errorset(rv);
 
-    return false;
+    goto exit_entry;
   }
   
   if ((request == NULL) || (request_len == 0) || (src == NULL)) {
     rv = EBUS_INVALID_PARA;
     bus_errorset(rv);
     
-		return false;
+    goto exit_entry;
   }
 
   data = net_mod_socket_svr_get(gNetmod_socket);
@@ -1577,14 +1717,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;
@@ -1602,7 +1735,7 @@
       
       logger->error("in BHReadRequest: Out of memory!\n");
       
-      return false;
+      goto exit_entry;
     }
     memset(topics_buf, 0x00, size + MIN_STR_LEN);
     
@@ -1620,28 +1753,40 @@
     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);
   }
 
-  bus_errorset(rv);
-	
+exit_entry:
+
+  if (rv != 0) { 
+    rrr.topic = STR_RSV;
+    rrr.data = STR_RSV;   
+
+    if ((proc_id != NULL) && (proc_id_len != NULL)) {
+
+      *proc_id_len = 0;
+      *proc_id = NULL;
+    }
+  }
+
+#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;
   
@@ -1730,3 +1875,22 @@
   return false;
 
 }
+
+int inter_key_get(void)
+{
+  if (gNetmod_socket != NULL)
+    return net_mod_socket_get_key(gNetmod_socket);
+
+  return SHM_BUS_KEY;
+}
+
+void *socket_data_get(void)
+{
+  return gNetmod_socket;
+}
+
+void inter_key_set(int key)
+{
+  net_mod_socket_bind(gNetmod_socket, key);
+}
+

--
Gitblit v1.8.0