固定bus key

wangzhengquan

2020-12-04 aa2f3b2a9968bb4928463bdae05fb026d16b60bb

 src/socket/shm_mod_socket.h

@@ -7,6 +7,7 @@
#include "hashtable.h"
#include "sem_util.h"
#include "logger_factory.h"
#include "key_def.h"
#include <set>

#define ACTION_LIDENTIFIER "<**"
@@ -14,8 +15,7 @@
#define TOPIC_LIDENTIFIER "{"
#define TOPIC_RIDENTIFIER "}"

static Logger *logger = LoggerFactory::getLogger();
#define BUS_MAP_KEY 1

//typedef std::basic_string<char, std::char_traits<char>, SHM_STL_Allocator<char> > SHMString;
typedef std::set<int,  std::less<int>, SHM_STL_Allocator<int> > SHMKeySet;
typedef std::map<SHMString, SHMKeySet *, std::less<SHMString>, SHM_STL_Allocator<std::pair<const SHMString, SHMKeySet *> > > SHMTopicSubMap;
@@ -41,17 +41,17 @@
   std::set<int> *bus_set;

private:
   inline int _recvfrom_(void **buf, int *size, int *port,  struct timespec *timeout, int flags);
   void _proxy_sub( char *topic, int port);
   void _proxy_pub( char *topic, size_t head_len, void *buf, size_t size, int port);
   inline int _recvfrom_(void **buf, int *size, int *key,  struct timespec *timeout, int flags);
   void _proxy_sub( char *topic, int key);
   void _proxy_pub( char *topic, size_t head_len, void *buf, size_t size, int key);
   void *run_pubsub_proxy();
   int parse_pubsub_topic(char *str, size_t size, char **_action, char **_topic, size_t *head_len );
   int _sub_( char *topic, int size, int port, struct timespec *timeout, int flags);
   int _pub_( char *topic, int topic_size, void *content, int content_size, int port, struct timespec *timeout, int flags);
   int _sub_( char *topic, int size, int key, struct timespec *timeout, int flags);
   int _pub_( char *topic, int topic_size, void *content, int content_size, int key, struct timespec *timeout, int flags);

   void _proxy_desub( char *topic, int port);
   void _proxy_desub_all(int port);
   int  _desub_( char *topic, int size, int port, struct timespec *timeout, int flags);
   void _proxy_desub( char *topic, int key);
   void _proxy_desub_all(int key);
   int  _desub_( char *topic, int size, int key, struct timespec *timeout, int flags);

   static void foreach_subscripters(std::function<void(SHMKeySet *, int)>  cb);
   static bool include_in_keys(int key, int keys[], size_t length);
@@ -62,48 +62,53 @@
   ShmModSocket();
   ~ShmModSocket();
    

   /**
    * 绑定端口到socket, 如果不绑定则系统自动分配一个
    * @return 0 成功， 其他值 失败的错误码
   */
   int bind(int port);
   int bind(int key);

   /**
    * 强制绑定端口到socket, 适用于程序非正常关闭的情况下，重启程序绑定原来还没释放的key
    * @return 0 成功， 其他值 失败的错误码
   */
   int force_bind(int port);
   int force_bind(int key);
   /**
    * 发送信息
    * @port 发送给谁
    * @key 发送给谁
    * @return 0 成功， 其他值 失败的错误码
    */
   int sendto(const void *buf, const int size, const int port);
   int sendto(const void *buf, const int size, const int key);
   // 发送信息超时返回。 @sec 秒 ， @nsec 纳秒
   int sendto_timeout(const void *buf, const int size, const int port, const struct timespec *timeout);
   int sendto_timeout(const void *buf, const int size, const int key, const struct timespec *timeout);
   // 发送信息立刻返回。
   int sendto_nowait(const void *buf, const int size, const int port);
   int sendto_nowait(const void *buf, const int size, const int key);

   /**
    * 接收信息
    * @port 从谁哪里收到的信息
    * @key 从谁哪里收到的信息
    * @return 0 成功， 其他值 失败的错误码
   */
   int recvfrom(void **buf, int *size, int *port);
   int recvfrom(void **buf, int *size, int *key);
   // 接受信息超时返回。 @sec 秒 ， @nsec 纳秒
   int recvfrom_timeout(void **buf, int *size, int *port,  struct timespec *timeout);
   int recvfrom_nowait(void **buf, int *size, int *port);
   int recvfrom_timeout(void **buf, int *size, int *key,  struct timespec *timeout);
   int recvfrom_nowait(void **buf, int *size, int *key);

   /**
    * 发送请求信息并等待接收应答
    * @port 发送给谁
    * @key 发送给谁
    * @return 0 成功， 其他值 失败的错误码
   */
   int sendandrecv(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) ;
   int sendandrecv(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) ;
   // 超时返回。 @sec 秒 ， @nsec 纳秒
   int sendandrecv_timeout(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size,  struct timespec *timeout) ;
   int sendandrecv_nowait(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) ;
   int sendandrecv_timeout(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size,  struct timespec *timeout) ;
   int sendandrecv_nowait(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) ;


   int sendandrecv_unsafe(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) ;
   // 超时返回。 @sec 秒 ， @nsec 纳秒
   int sendandrecv_unsafe_timeout(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size,  struct timespec *timeout) ;
   int sendandrecv_unsafe_nowait(const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) ;


   /**
@@ -117,41 +122,41 @@
    * 订阅指定主题
    * @topic 主题
    * @size 主题长度
    * @port 总线端口
    * @key 总线端口
    */
   int  sub(char *topic, int size, int port);
   int  sub(char *topic, int size, int key);
   // 超时返回。 @sec 秒 ， @nsec 纳秒
   int  sub_timeout(char *topic, int size, int port,  struct timespec *timeout);
   int  sub_nowait(char *topic, int size, int port);
   int  sub_timeout(char *topic, int size, int key,  struct timespec *timeout);
   int  sub_nowait(char *topic, int size, int key);


    /**
    * 取消订阅指定主题
     * @topic 主题,主题为空时取消全部订阅
    * @size 主题长度
    * @port 总线端口
    * @key 总线端口
    */
   int desub( char *topic, int size, int port);
   int desub( char *topic, int size, int key);
   // 超时返回。 @sec 秒 ， @nsec 纳秒
   int desub_timeout(char *topic, int size, int port, struct timespec *timeout);
   int desub_nowait(char *topic, int size, int port) ;
   int desub_timeout(char *topic, int size, int key, struct timespec *timeout);
   int desub_nowait(char *topic, int size, int key) ;

   /**
    * 发布主题
    * @topic 主题
    * @content 主题内容
    * @port 总线端口
    * @key 总线端口
    */
   int  pub(char *topic, int topic_size, void *content, int content_size, int port);
   int  pub(char *topic, int topic_size, void *content, int content_size, int key);
   //  超时返回。 @sec 秒 ， @nsec 纳秒
   int  pub_timeout(char *topic, int topic_size, void *content, int content_size, int port,  struct timespec *timeout);
   int  pub_nowait(char *topic, int topic_size, void *content, int content_size, int port);
   int  pub_timeout(char *topic, int topic_size, void *content, int content_size, int key,  struct timespec *timeout);
   int  pub_nowait(char *topic, int topic_size, void *content, int content_size, int key);


   /**
    * 获取soket端口号
    * 获取soket key
    */
   int get_port() ;
   int get_key() ;


};