#ifndef _BUS_SERVER_SOCKET_H_ #define _BUS_SERVER_SOCKET_H_ #include "usg_common.h" #include "shm_socket.h" #include "shm_allocator.h" #include "shm_mm.h" #include "hashtable.h" #include "sem_util.h" #include "logger_factory.h" #include "key_def.h" #include "socket_def.h" #include //typedef std::basic_string, SHM_STL_Allocator > SHMString; typedef std::set, SHM_STL_Allocator > SHMKeySet; typedef std::map, SHM_STL_Allocator > > SHMTopicSubMap; class BusServerSocket { private: shm_socket_t *shm_socket; // pthread_t recv_thread; // <主题, 订阅者> SHMTopicSubMap *topic_sub_map; private: int destroy(); void _proxy_sub( char *topic, int key); void _proxy_pub( char *topic, char *buf, size_t size, int key); int _run_proxy_(); // int parse_pubsub_topic(char *str, size_t size, char **_action, char **_topic, size_t *head_len ); void _proxy_desub( char *topic, int key); void _proxy_desub_all(int key); void _proxy_reg(const char *topic, size_t topic_size, const char *content, size_t content_size, int key, int flag); static void foreach_subscripters(std::function cb); // static bool include_in_keys(int key, int keys[], size_t length); public: static size_t remove_subscripters(int keys[], size_t length) ; public: BusServerSocket(); ~BusServerSocket(); /** * 绑定端口到socket, 如果不绑定则系统自动分配一个 * @return 0 成功, 其他值 失败的错误码 */ int bind(int key); /** * 强制绑定端口到socket, 适用于程序非正常关闭的情况下,重启程序绑定原来还没释放的key * @return 0 成功, 其他值 失败的错误码 */ int force_bind(int key); /** * 启动bus * * @return 0 成功, 其他值 失败的错误码 */ int start(); /** * 停止bus * * @return 0 成功, 其他值 失败的错误码 */ int stop(); /** * 获取soket key */ int get_key() ; void _data_remove(int val1, int val2); }; #endif