bug fixed ps_sub thread join

zhangmeng

2023-03-01 777333ff834744ac5665fa9abe5ec6373d25cda8

 message.h

@@ -3,42 +3,20 @@

#include <stddef.h>

struct cstr{
    char*  str;
    size_t size;
};
// 进程信息
struct cproc{
    struct cstr name;
    struct cstr id;
    struct cstr info;
};
/*
    注册进程的proc信息，包括进程名[name], 进程[id],等
*/
typedef struct cproc cproc;
/*
    进程注册信息, 包括进程的proc信息 cproc
    待注册的本进程提供请求响应的主题 topics[s], char** rep, 主题数量 repcnt
    待注册的本进程发布的主题 topics[s], char** pub, 主题数量 pubcnt
    待注册的本进程订阅的主题 topics[s], char** sub, 主题数量 subcnt
    待注册的本进程订阅的[网络、远程主机]主题 topics[s], char** subnet, 主题数量 subnetcnt
*/
typedef struct creg creg;

struct cstrarr{
    struct cstr* arr;
    size_t count;
};
// 进程注册信息
struct creg{
    struct cproc*  pinfo;

    cstrarr channel;
    cstrarr topic_pub;
    cstrarr topic_sub;
    cstrarr topic_sub_net;
};

// 其他进程信息
struct cclient{
    struct creg*  rinfo;

    int replykey;
    int hbcnt;
    int dcnt;
    int status;
};

//TableChanged enum
//TableChanged enum pollcontrol 接收到的订阅消息，数据库的变化
enum TableChanged {
    T_Camera = 0,       //摄像机变化
    T_CameraRule = 1,       //摄像机任务参数变化
@@ -58,121 +36,265 @@
    T_CameraPolygonRelation = 15,        //摄像机区域的关联关系
    T_Voice = 16,       //报警声音发生变化
};
// pollcontrol 接收到的订阅消息，数据库的操作
enum DbAction {
    Insert = 0,     //Insert
    Update = 1,     //Update
    Delete = 2,     //Delete
    Insert = 0,     //Insert db
    Update = 1,     //Update db
    Delete = 2,     //Delete db
};

// 数据库变化信息
struct DbChangeMsg{
    TableChanged table;     //变化的表
    struct cstr   id;      //变化数据id
    DbAction action;        //action
    struct cstr   info;    //变化内容
    TableChanged    table;     //变化的表
    char*           id;      //变化数据id
    size_t          idl;     // id len
    DbAction action;        //action/ DbAction[Insert/Update/Delete]
    char*           info;    //变化内容
    size_t          infol;
};

// 订阅消息
// 订阅消息, 包括数据库消息和其他进程的消息
struct csubmsg{
    struct cstr topic;
    char* topic;      // 收到的订阅消息的主题，区分那种订阅消息
    size_t topicl;

    // private
    // enum MsgT {NONE=0, DB, PROCLIST} type;
    struct cstr msg;
    char* msg;        // 收到的订阅消息的body
    size_t msgl;
};

// 其他进程信息, pollcontrol可能依赖其他进程的服务，可能需要等其他进程启动再开始工作
struct cclient{
    creg*  rinfo;               // 代表其他进程的进程信息

    int replykey;               // 没用，上一个版本用共享内存，此为key
    int hbcnt;                  // 心跳次数？可能没用
    int dcnt;                   // deadcount，可能没用？
    int status;                 // 进程状态，可能没用？
};

// 其他注册进程列表，如数据库进程需要启动再运行pollcontrol逻辑
struct cproclist{
    struct  cclient* cli;
    size_t  clientsize;
    struct  cclient* cli;   // 其他进程的列表
    size_t  count;          // 其他进程的列表的count
};

//////////////////////////////////////////
// request msg
/*
    此消息既作为接收的request msg的结构，也作为发送request 消息的结构
    接收request消息会带有发出request消息的进程的id，procid
*/
struct creqmsg{
    struct cstr procid;
    struct cstr msg;
    char* procid;     // 发送request消息的进程id
    size_t procidl;
    char* msg;        // request消息体
    size_t msgl;
    char* path;
    size_t pathl;
    char* body;
    size_t bodyl;
};
// decode stack err msg
/*
    pollcontrol会接收数据栈解码发送的解码失败消息，用于置网页状态
*/
struct cstackmsgerr{
    struct cstr stackid;
    struct cstr fileid;
    char* stackid;    // 解码失败的数据栈id
    size_t stackidl;
    char* fileid;     // 解码失败的文件id
    size_t fileidl;
};
// stack file
/*
    数据栈的文件列表，对应 protomsg.FileAnalysis 但是此结构体非常长
    看代码应该只需要解析出的几个
*/
struct cstackfile{
    struct cstr id;
    struct cstr name;
    struct cstr path;
    int type;
    void* noused;
    char* id;         // 文件id
    size_t idl;
    char* name;       // 文件name
    size_t namel;
    char* path;       // 文件路径
    size_t pathl;

    int type;               // 文件类型 1：video，2：picture
    void* noused;           // 未使用
};
// decode stack success msg
/*
    pollcontrol会接收数据栈解码完成的消息，然后下发新的数据栈任务
*/
struct cstackmsg{
    int procnum;
    struct cstr stackid;
    struct cstr stackname;
    int type;
    int shmkey;
    int width;
    int procnum;                // decoder 启动的进程号，数据栈可能会有数个decoder同时运行
    char* stackid;        // 数据栈 id
    size_t stackidl;
    char* stackname;      // 数据栈 name
    size_t stacknamel;
    int type;                   // 数据栈类型 video picture
    int shmkey;                 // 数据栈使用的共享内存key
    int width;                  // 分辨率
    int height;
    int fps;
    struct cstackfile* files;
    int fps;                    // fps 采样 fps/s
    struct cstackfile* files;   // 数据栈文件列表
    size_t filescnt;
};

// reply msg
// 对应 bhome_msg.MsgRequestTopicReply
struct crepmsg{
    int errcode;
    struct cstr errmsg;
    struct cstr data;
    int errcode;                // 相应request请求的消息，错误码
    char* errmsg;         // 错误消息
    size_t errmsgl;
    char* data;           // 消息体
    size_t datal;
};

/*
    对应 bhome_msg.MsgQueryProcReply_Info query procs 返回值
*/
struct cqueryprocs{
    char* id;
    size_t idl;
    int online;
};

#ifdef __cplusplus
extern "C"{
#endif

struct cstr cstr_new(const char* str, const size_t len);
void cstr_free(struct cstr str);
struct cstrarr cstr_arr_new(const size_t count);
void cstr_arr_add(struct cstrarr* arr, const char* data, const size_t len, const size_t idx);
void cstr_arr_free(struct cstrarr arr);
/*
    内部使用，深拷贝 cproc
*/
cproc* internal_clone_cproc(const cproc* proc);
void internal_cproc_free(cproc* proc);
/*
    返回一个 cproc 结构，仅拷贝输入参数
    生命周期 cproc < name[id]
*/
cproc* make_cproc(const char* name, const char* id);
/*
    从已存在的 cproc 中获取 name 和 id，仅仅是 cproc 内部的引用，无需释放
    生命周期 name[id] < cproc
*/
char* cproc_name(const cproc* proc);
char* cproc_id(const cproc* proc);
/*
    释放 cproc 指针
*/
void cproc_free(cproc* proc);

void free_creg(struct creg* reg);

struct cproc* clone_proc_info(struct cproc* pi);
void free_proc_info(struct cproc* pi);
/*
    返回一个 creg 结构，仅拷贝输入参数
    输入参数生命周期需覆盖返回值生命周期
*/
creg* make_creg(const cproc* proc, const char** rep, const size_t repcnt,
                                    const char** pub, const size_t pubcnt,
                                    const char** sub, const size_t subcnt,
                                    const char** subnet, const size_t subnetcnt);
creg* make_creg_from_cproc(const cproc* proc);
void creg_add_topic_reply(creg* reg, const char** topic, const size_t count);
void creg_add_topic_pub(creg* reg, const char** topic, const size_t count);
void creg_add_topic_sub(creg* reg, const char** topic, const size_t count);
void creg_add_topic_subnet(creg* reg, const char** topic, const size_t count);
/*
    获取已存在的 creg 中的 cproc 或者注册的主题，无需释放
    生命周期 < creg
*/
const cproc* creg_proc(const creg* reg);
char** creg_reply_topic(const creg* reg, size_t* count);
char** creg_pub_topic(const creg* reg, size_t* count);
char** creg_sub_topic(const creg* reg, size_t* count);
char** creg_subnet_topic(const creg* reg, size_t* count);
/*
    释放 creg 指针
*/
void creg_free(creg* reg);

// 订阅消息相关，订阅数据库db消息和进程列表proclist消息
/*
    cbhomeclient.cpp中使用，将接收到的submsg解包成csubmsg
    包括topic和msg数据，msg数据并未反序列化
    必须使用free_submsg释放
*/
struct csubmsg* to_submsg(const char* data, const size_t size);
void free_submsg(struct csubmsg* msg);
// db msg
/*
    接收到的submsg在pollcontrol中可能为数据库更新的通知
    将csubmsg中未反序列化的数据解包成db的数据格式
    必须用free_submsg_db释放
*/
struct DbChangeMsg* get_submsg_db(struct csubmsg* msg);
void free_submsg_db(struct DbChangeMsg* msg);
// proclist msg
/*
    接收到的submsg在pollcontrol中可能为关心的其他进程的列表
    将csubmsg中未反序列化的数据解包成进程的列表
    必须用free_submsg_proclist释放
*/
struct cproclist* get_submsg_proclist(struct csubmsg* msg);
void free_submsg_proclist(struct cproclist* ppl);

// request msg
/*
    cbhomeclient.cpp中使用，将接收到的request请求数据解包成creqmsg
    包括request进程的procid和msg数据，msg数据并未反序列化
    必须使用free_reqmsg释放
*/
struct creqmsg* to_reqmsg(const char* pid,const size_t pids,const char* data,const size_t size);
struct creqmsg* make_req_msg(const char* topic, const size_t topics,
void free_reqmsg(struct creqmsg* msg);
/*
    本进程向其他进程请求数据构建
    包括请求的主题topic和请求的数据body[data]
    必须使用free_reqmsg释放
*/
struct creqmsg* make_req_msg(const char* topic, const size_t topicl,
                                const char* data, const size_t datal);

void free_reqmsg(struct creqmsg* msg);
// decode err msg
/*
    接收到的creqmsg在pollcontrol中可能为数据栈解码错误信息
    将creqmsg中未反序列化的数据解包成cstackmsgerr，包括数据栈id stackid和文件id fileid
    必须用free_reqmsg_stackerr释放
*/
struct cstackmsgerr* get_reqmsg_stackerr(struct creqmsg* msg);
void free_reqmsg_stackerr(struct cstackmsgerr* msg);
// decode success msg
/*
    接收到的creqmsg在pollcontrol中可能为数据栈解码完成信息
    将creqmsg中未反序列化的数据解包成cstackmsg
    包括
    int procnum;    // 启动的decoder进程num
    struct cstr stackid;
    struct cstr stackname;
    int type;   // video picture
    int shmkey;
    int width;
    int height;
    int fps;
    struct cstackfile* files; // 文件列表
    size_t filescnt; // 文件数量
    必须用free_reqmsg_stackerr释放
*/
struct cstackmsg* get_reqmsg_stack(struct creqmsg* msg);
void free_reqmsg_stack(struct cstackmsg* msg);

// reply msg
struct cstr make_reply_msg_json(const int success, const char* msg, const size_t msgl,
                                const char* data, const size_t datal);
/*
    创建 creqmsg 包括errcode、errmsg和消息体data
    必须使用 free_reply_msg 释放
*/
struct crepmsg* make_reply_msg(const int errcode, const char* errmsg, const size_t emsgl,
                                const char* data, const size_t datal);

void free_reply_msg(struct crepmsg* msg);

/*
    释放 query procs 从center获取所有的procs信息
*/
void free_query_procs(struct cqueryprocs* procs, const size_t count);

#ifdef __cplusplus
}
#endif