From 68d23225a38a35f1325eb39fa4ed5a005d5de473 Mon Sep 17 00:00:00 2001 From: fujuntang <fujuntang@aiot.com> Date: 星期三, 11 八月 2021 09:50:20 +0800 Subject: [PATCH] fix from 3.1 first commit --- src/socket/bus_server_socket.cpp | 106 ++++++++++++++++++++++++++++------------------------ 1 files changed, 57 insertions(+), 49 deletions(-) diff --git a/src/socket/bus_server_socket.cpp b/src/socket/bus_server_socket.cpp index 462fd9d..7a45696 100644 --- a/src/socket/bus_server_socket.cpp +++ b/src/socket/bus_server_socket.cpp @@ -6,7 +6,7 @@ static Logger *logger = LoggerFactory::getLogger(); void BusServerSocket::foreach_subscripters(std::function<void(SHMKeySet *, int)> cb) { - SHMTopicSubMap *topic_sub_map = mem_pool_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); + SHMTopicSubMap *topic_sub_map = shm_mm_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); SHMKeySet *subscripter_set; SHMKeySet::iterator set_iter; SHMTopicSubMap::iterator map_iter; @@ -29,7 +29,7 @@ int key; for(int i = 0; i < length; i++) { key = keys[i]; - SHMTopicSubMap *topic_sub_map = mem_pool_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); + SHMTopicSubMap *topic_sub_map = shm_mm_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); SHMKeySet *subscripter_set; SHMKeySet::iterator set_iter; SHMTopicSubMap::iterator map_iter; @@ -39,7 +39,6 @@ subscripter_set = map_iter->second; if(subscripter_set != NULL && (set_iter = subscripter_set->find(key) ) != subscripter_set->end()) { subscripter_set->erase(set_iter); -// printf("remove_subscripter %s, %d\n", map_iter->first, key); count++; } } @@ -52,31 +51,13 @@ BusServerSocket::BusServerSocket() { logger->debug("BusServerSocket Init"); - shm_socket = shm_open_socket(SHM_SOCKET_DGRAM); + shm_socket = shm_socket_open(SHM_SOCKET_DGRAM); topic_sub_map = NULL; } BusServerSocket::~BusServerSocket() { - SHMKeySet *subscripter_set; - SHMTopicSubMap::iterator map_iter; - - stop(); - - if(topic_sub_map != NULL) { - for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) { - subscripter_set = map_iter->second; - if(subscripter_set != NULL) { - subscripter_set->clear(); - mm_free((void *)subscripter_set); - } - - } - topic_sub_map->clear(); - mem_pool_free_by_key(SHM_BUS_MAP_KEY); - } - shm_close_socket(shm_socket); - logger->debug("BusServerSocket destory 3"); + destroy(); } @@ -97,12 +78,11 @@ * 鍚姩bus * * @return 0 鎴愬姛锛� 鍏朵粬鍊� 澶辫触鐨勯敊璇爜 -*/ + */ int BusServerSocket::start(){ - topic_sub_map = mem_pool_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); + topic_sub_map = shm_mm_attach<SHMTopicSubMap>(SHM_BUS_MAP_KEY); - run_pubsub_proxy(); - // 杩涚▼鍋滄鐨勬椂鍊欙紝棰勭暀3绉掕祫婧愬洖鏀剁殑鏃堕棿銆傚惁鍒欙紝浼氬彂鐢熻皟鐢╟lose鐨勬椂鍊欙紝鍏变韩鍐呭瓨鐨勮祫婧愯繕娌℃潵寰楀強鍥炴敹杩涚▼灏遍��鍑轰簡 + _run_proxy_(); return 0; } @@ -112,8 +92,6 @@ if( shm_socket->key <= 0) { return -1; } - // snprintf(buf, 128, "%sstop%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, "", TOPIC_RIDENTIFIER); - // return shm_sendto(shm_socket, buf, strlen(buf), shm_socket->key, NULL, 0); bus_head_t head = {}; memcpy(head.action, "stop", sizeof(head.action)); head.topic_size = 0; @@ -123,13 +101,33 @@ void *buf; int size = ShmModSocket::get_bus_sendbuf(head, NULL, 0, NULL, 0, &buf); if(size > 0) { - ret = client.sendandrecv_unsafe( buf, size, shm_socket->key, NULL, NULL); + ret = client.sendto( buf, size, shm_socket->key); free(buf); return ret; } else { return -1; } +} + +int BusServerSocket::destroy() { + SHMKeySet *subscripter_set; + SHMTopicSubMap::iterator map_iter; + if(topic_sub_map != NULL) { + for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) { + subscripter_set = map_iter->second; + if(subscripter_set != NULL) { + subscripter_set->clear(); + mm_free((void *)subscripter_set); + } + + } + topic_sub_map->clear(); + shm_mm_free_by_key(SHM_BUS_MAP_KEY); + } + shm_socket_close(shm_socket); + logger->debug("BusServerSocket destory 3"); + return 0; } /* @@ -194,21 +192,20 @@ std::vector<int>::iterator vector_iter; int send_key; + int rv; struct timespec timeout = {1,0}; if( (map_iter = topic_sub_map->find(topic) ) != topic_sub_map->end()) { + subscripter_set = map_iter->second; for(set_iter = subscripter_set->begin(); set_iter != subscripter_set->end(); set_iter++) { send_key = *set_iter; - // printf("_proxy_pub send before %d \n", send_key); - if (shm_sendto(shm_socket, buf, size, send_key, &timeout) == EBUS_CLOSED ) { - //瀵规柟宸插叧闂殑杩炴帴鏀惧埌寰呭垹闄ら槦鍒楅噷銆傚鏋滅洿鎺ュ垹闄や細璁﹊ter鎸囬拡鍑虹幇閿欎贡 - subscripter_to_del.push_back(send_key); - } else { -// printf("_proxy_pub send after: %d \n", send_key); + rv = shm_sendto(shm_socket, buf, size, send_key, &timeout, BUS_TIMEOUT_FLAG); + if(rv == 0) { + continue; } - - + //瀵规柟宸插叧闂殑鎴栬�呭搴旂殑杩涚▼琚玨ill鎺夌殑杩炴帴鏀惧埌寰呭垹闄ら槦鍒楅噷銆傚鏋滅洿鎺ュ垹闄や細璁﹊ter鎸囬拡鍑虹幇閿欎贡 + subscripter_to_del.push_back(send_key); } // 鍒犻櫎宸插叧闂殑绔� @@ -223,8 +220,9 @@ } } -void * BusServerSocket::run_pubsub_proxy() { - // pthread_detach(pthread_self()); + +// 杩愯浠g悊 +void * BusServerSocket::_run_proxy_() { int size; int key; char * action, *topic, *topics, *buf, *content; @@ -232,26 +230,28 @@ char resp_buf[128]; bus_head_t head; + int rv; + char send_buf[512] = { 0x00 }; + const char *topic_delim = ","; -// printf("run_pubsub_proxy server receive before\n"); + while(shm_recvfrom(shm_socket, (void **)&buf, &size, &key) == 0) { -// printf("run_pubsub_proxy server recvfrom %d after: %s \n", key, buf); head = ShmModSocket::decode_bus_head(buf); topics = buf + BUS_HEAD_SIZE; action = head.action; - // printf("run_pubsub_proxy : %s\n", action); + if(strcmp(action, "sub") == 0) { // 璁㈤槄鏀寔澶氫富棰樿闃� topic = strtok(topics, topic_delim); -// printf("run_pubsub_proxy topic = %s\n", topic); while(topic) { + _proxy_sub(trim(topic, 0), key); topic = strtok(NULL, topic_delim); } } else if(strcmp(action, "desub") == 0) { -// printf("desub topic=%s,%s,%d\n", topics, trim(topics, 0), strcmp(trim(topics, 0), "")); + if(strcmp(trim(topics, 0), "") == 0) { // 鍙栨秷鎵�鏈夎闃� _proxy_desub_all(key); @@ -259,6 +259,7 @@ topic = strtok(topics, topic_delim); while(topic) { + _proxy_desub(trim(topic, 0), key); topic = strtok(NULL, topic_delim); } @@ -270,18 +271,25 @@ _proxy_pub(topics, content, head.content_size, key); } - else if(strcmp(action, "stop") == 0) { + else if (strncmp(buf, "request", strlen("request")) == 0) { + sprintf(send_buf, "%4d", key); + strncpy(send_buf + 4, buf, (sizeof(send_buf) - 4) >= (strlen(buf) + 1) ? strlen(buf) : (sizeof(send_buf) - 4)); + + rv = shm_sendto(shm_socket, send_buf, strlen(send_buf) + 1, key); + if(rv != 0) { + logger->error( "BusServerSocket::_run_proxy_ : requst answer fail!\n"); + } + } + else if(strcmp(action, "stop") == 0) { logger->info( "Stopping Bus..."); free(buf); break; } else { - logger->error( "BusServerSocket::run_pubsub_proxy : unrecognized action %s", action); + logger->error( "BusServerSocket::_run_proxy_ : unrecognized action %s", action); } free(buf); } - - shm_sendto(shm_socket, "stop_finished", strlen( "stop_finished") +1, key); return NULL; } -- Gitblit v1.8.0