增加BHSubscribeNetTopics、BHQueryTopicAddress接口 修改CMakeLists.txt，增加相应文件到编译中。(需...

shenxin

2021-08-23 08505499d1449f53449e119aefc71324376b348b

增加BHSubscribeNetTopics、BHQueryTopicAddress接口
修改CMakeLists.txt，增加相应文件到编译中。(需要安装protobuf)

 CMakeLists.txt

@@ -19,7 +19,7 @@


list(APPEND EXTRA_INCLUDES "${PROJECT_SOURCE_DIR}/include/usgcommon")
list(APPEND EXTRA_LIBS ${PROJECT_SOURCE_DIR}/lib/libusgcommon.a  pthread rt)
list(APPEND EXTRA_LIBS ${PROJECT_SOURCE_DIR}/lib/libusgcommon.a  pthread rt protobuf-lite.a)

# build api doc
if (BUILD_DOC)

 src/CMakeLists.txt

@@ -26,6 +26,10 @@
./shm/mm.cpp
./shm/hashtable.cpp
./shm/shm_mm.cpp
./bh_api.cc
./proto/bhome_msg.pb.cc
./proto/bhome_msg_api.pb.cc
./proto/error_msg.pb.cc

)

@@ -87,6 +91,7 @@
./shm/shm_mm_wrapper.h
./shm/shm_allocator.h
./shm/shm_mm.h
./bh_api.h


  DESTINATION include)

 src/bh_api.cc

@@ -4,6 +4,8 @@
#include "proto/bhome_msg.pb.h"
#include "proto/bhome_msg_api.pb.h"

using namespace bhome_msg;

int BHRegister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms)
{
    //反序列化入参
@@ -151,6 +153,69 @@
    return errCode==0?true: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)
{
    //反序列化入参
    BHAddress input0;
    MsgQueryTopic input1;
    if (!input0.ParseFromArray(remote,remote_len)||!input1.ParseFromArray(topic,topic_len))
        return false;
    struct _BHAddress
    {
        unsigned long long mq_id;
        long long abs_addr;
        const char *ip;
        int port;
    }_input0;
    const char *_input1;
    _input0.mq_id=input0.mq_id();
    _input0.abs_addr=input0.abs_addr();
    _input0.ip=input0.ip().c_str();
    _input0.port=input0.port();
    _input1=input1.topic().c_str();

    //调用函数实现
    int errCode=0;//0 is success
    //此种以固定结构返回数据的方式,只能暂时固定数量上限。这里限制最多128个
    //若不限数量,需重新设计数据返回的方式
    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;//实际在mtr_list中填写了多少个项目
    //errCode=call_QueryTopicAddress(&_input0,_input1,mtr_list,&mtr_list_num,timeout_ms);//这句是功能的具体实现
    char errString[32]={0};
    if(errCode!=0)
        strcpy(errString,"错误原因描述...");

    //序列化返回数据
    MsgQueryTopicReply mtr;
    mtr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(errCode));
    mtr.mutable_errmsg()->set_errstring(errString);
    for(int i=0;i<mtr_list_num;i++)
    {
        MsgQueryTopicReply_BHNodeAddress *mtrb=mtr.add_node_address();
        mtrb->set_proc_id(mtr_list[i].proc_id);
        mtrb->mutable_addr()->set_mq_id(mtr_list[i].mq_id);
        mtrb->mutable_addr()->set_abs_addr(mtr_list[i].abs_addr);
        mtrb->mutable_addr()->set_ip(mtr_list[i].ip);
        mtrb->mutable_addr()->set_port(mtr_list[i].port);
    }
    *reply_len=mtr.ByteSizeLong();
    *reply=malloc(*reply_len);
    mtr.SerializePartialToArray(*reply,*reply_len);

    return errCode==0?true:false;
}

int BHQueryProcs(const void *remote,
                 const int remote_len,
                 const void *query,
@@ -226,6 +291,41 @@

int BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms)
{
    //反序列化入参
    MsgTopicList input;
    if(!input.ParseFromArray(topics,topics_len))
        return false;
    struct _MsgTopicList
    {
        int amount;
        const char *topics[128];
    }_input;
    _input.amount=input.topic_list_size();
    for(int i=0;i<_input.amount;i++)
        _input.topics[i]=input.topic_list(i).c_str();

    //调用函数实现
    int errCode=0;//0 is success
    //errCode=call_SubscribeTopics(&_input,timeout_ms);//这句是功能的具体实现
    char errString[32]={0};
    if(errCode!=0)
        strcpy(errString,"错误原因描述...");

    //序列化返回数据
    MsgCommonReply mcr;
    mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(errCode));
    mcr.mutable_errmsg()->set_errstring(errString);
    *reply_len=mcr.ByteSizeLong();
    *reply=malloc(*reply_len);
    mcr.SerializePartialToArray(*reply,*reply_len);

    return errCode==0?true:false;
}

int BHSubscribeNetTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms)
{
    //same as BHSubscribeTopics
	
    //反序列化入参
    MsgTopicList input;
    if(!input.ParseFromArray(topics,topics_len))
@@ -348,17 +448,17 @@

    //调用函数实现
    int errCode=0;//0 is success
    std::string msg_id;
    //errCode=call_AsyncRequest(&_input0,&_input1,&msg_id,&timeout_ms);//这句是功能的具体实现
    std::string MsgID;
    //errCode=call_AsyncRequest(&_input0,&_input1,&MsgID,&timeout_ms);//这句是功能的具体实现
    if(errCode!=0)
        return false;

    //序列化返回数据
    if(msg_id==nullptr||msg_id_len==nullptr)//这里有可能调用者不需要返回
        return true;
    *msg_id_len=msg_id.size();
    *msg_id_len=MsgID.size();
    *msg_id=malloc(*msg_id_len);
    memcpy(*msg_id,msg_id.data(),*msg_id_len);
    memcpy(*msg_id,MsgID.data(),*msg_id_len);

    return true;
}

 src/bh_api.h

@@ -23,6 +23,11 @@
                     int *reply_len,
                     const int timeout_ms);

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);

int BHQueryProcs(const void *remote,
                 const int remote_len,
                 const void *query,
@@ -30,6 +35,17 @@
                 void **reply,
                 int *reply_len,
                 const int timeout_ms);

int BHSubscribeTopics(const void *topics,
                      const int topics_len,
                      void **reply,
                      int *reply_len,
                      const int timeout_ms);
int BHSubscribeNetTopics(const void *topics,
                         const int topics_len,
                         void **reply,
                         int *reply_len,
                         const int timeout_ms);

int BHHeartbeatEasy(const int timeout_ms);
int BHHeartbeat(const void *proc_info,
@@ -79,8 +95,6 @@
int BHCleanup();

void BHFree(void *buf, int size);

int BHGetLastError(void **msg, int *msg_len);

#ifdef __cplusplus
}