总述
设计模式有很多，发布与订阅模式便是其中之一，具有很好的解耦作用，程序开发量一大，如果程序框架没有设计好，杂糅在一起，后期的维护成本将是巨大的。
模式介绍
发布与订阅模式，由一个消息代理中心基于消息主题进行数据的传输与转发，完全隔离发送者与接收者之间的联系，实现高度的解耦，当然该模式也存在缺点，例如由于存在消息代理，所以效率没有直接传输的高等。
C语言实现
/**
* @file msg_com.h
* @author your name (you@domain.com)
* @brief pub and sub communication mode
* @version 0.1
* @date 2023-05-28
*
* @copyright Copyright (c) 2023
*
*/
#ifndef MSG_COM_H
#define MSG_COM_H
#include
#include
#include
#include
#include "log.h"
#ifdef __cplusplus
extern "C"
{
#endif
#define MSG_CENTER_LOG(fmt, ...) COMLOG_DEBUG_LOG(fmt, ##__VA_ARGS__)
#define ANY_ID (0xFFFFFFFF)
typedef void(*HandleEvent)(char *event_name, void *data, int srcId);
typedef void(*MutexLock)(bool lock);
/* 订阅事件节点 */
typedef struct EventSubscriberNode
{
char event[32];
HandleEvent func;
int needId;
struct EventSubscriberNode *next;
}EventSubscriberNode_t;
/* 发布事件节点 */
typedef struct EventPublishNode
{
char event[32];
void *data; //内部会拷贝一份，并自动释放
int srcId;
struct EventPublishNode *next;
}EventPublishNode_t;
typedef struct MsgCenterHandle
{
MutexLock mutexSubscriber;
MutexLock mutexPublish;
EventSubscriberNode_t *eventSubscriberHead;
EventPublishNode_t *eventPublishHead;
volatile int exit_flg;
}MsgCenterHandle_t;
/**
* @brief init msg center
*
* @param mutexSubscriber mutex lock func
* @param mutexPublish mutex lock func
* @return MsgCenterHandle_t* handle
*/
MsgCenterHandle_t *MsgCenterInit(MutexLock mutexSubscriber, MutexLock mutexPublish);
/**
* @brief deinit handle and set it null
*
* @param msg_handle MsgCenterHandle_t **
*/
void MsgCenterDeinit(MsgCenterHandle_t **msg_handle);
/**
* @brief loop call this func,
* the shorter the interval, the faster the event execution
* @param msg_handle MsgCenterHandle_t *
* @return int -1 error 0 success
*/
int MsgCenterLoop(MsgCenterHandle_t *msg_handle);
/**
* @brief Subscriber one event
*
* @param handle
* @param event event topic name
* @param func
* @param needId only listen the event with the needId(Event sender ID),
* ANY_ID not judged needId(Event sender ID)
* @return int -1 error 0 success
*/
int SubscriberEvent(MsgCenterHandle_t *handle, char *event, HandleEvent func, int needId);
/**
* @brief cancel listen one event
*
* @param handle
* @param event
* @param func
* @return int -1 error 0 success
*/
int ReleaseSubscriberEvent(MsgCenterHandle_t *handle, char *event, HandleEvent func);
/**
* @brief publish one event
*
* @param handle
* @param event event topic name
* @param data
* @param datalen
* @param srcId Event sender ID
* @return int -1 error 0 success
*/
int PublishEvent(MsgCenterHandle_t *handle, char *event, void *data, int datalen, int srcId);
#ifdef __cplusplus
}
#endif
#endif
/**
* @file msg_com.c
* @author your name (you@domain.com)
* @brief pub and sub communication mode
* @version 0.1
* @date 2023-05-28
*
* @copyright Copyright (c) 2023
*
*/
#include "msg_com.h"
static void MsgMutexLock(MutexLock lock, bool is_lock)
{
if(lock)
{
lock(is_lock);
}
}
int MsgCenterLoop(MsgCenterHandle_t *msg_handle)
{
//MSG_CENTER_LOG("EventProcess start \r\n");
MsgCenterHandle_t *handle = (MsgCenterHandle_t *)msg_handle;
if(handle == NULL)
{
MSG_CENTER_LOG("handle is null");
return -1;
}
if (!handle->exit_flg)
{
EventPublishNode_t cur = {0};
/*find last valid publish event*/
MsgMutexLock(handle->mutexPublish, true);
EventPublishNode_t **curPublish = &handle->eventPublishHead;
EventPublishNode_t *tempPublish = NULL;
/*curPublish = last_node->next pointer address, so modify *curPublish = modify last_node->next obj*/
while ((*curPublish) != NULL )
{
if(strlen((*curPublish)->event) > 0)
{
if((*curPublish)->next == NULL)
{
memcpy(&cur, *curPublish, sizeof(EventPublishNode_t));
break;
}
memcpy(&cur, *curPublish, sizeof(EventPublishNode_t));
/*save the next node of valid cur node*/
tempPublish = (*curPublish)->next;
break;
}
curPublish = &((*curPublish)->next);
}
if(curPublish != NULL && *curPublish != NULL)
{
memset((*curPublish)->event, 0, sizeof((*curPublish)->event));
free((*curPublish));
*curPublish = NULL;
//delete cur node, and use the next node replace
if(tempPublish)
{
*curPublish = tempPublish;
}
}
MsgMutexLock(handle->mutexPublish, false);
if (strlen(cur.event) <= 0)
{
return 0;
}
MsgMutexLock(handle->mutexSubscriber, true);
EventSubscriberNode_t *curSubscriber = (EventSubscriberNode_t *)handle->eventSubscriberHead;
while (curSubscriber != NULL)
{
if(curSubscriber->needId == cur.srcId && strcmp(curSubscriber->event , cur.event) == 0)
{
curSubscriber->func(cur.event, cur.data, cur.srcId);
}
else if(curSubscriber->needId == ANY_ID && strcmp(curSubscriber->event , cur.event) == 0)
{
curSubscriber->func(cur.event, cur.data, ANY_ID);
}
curSubscriber = curSubscriber->next;
}
MsgMutexLock(handle->mutexSubscriber, false);
if(cur.data)
{
free(cur.data);
}
}
//MSG_CENTER_LOG("EventProcess end \r\n");
return 0;
}
MsgCenterHandle_t *MsgCenterInit(MutexLock mutexSubscriber, MutexLock mutexPublish)
{
MsgCenterHandle_t *handle = (MsgCenterHandle_t *)malloc(sizeof(MsgCenterHandle_t));
if(handle == NULL)
{
MSG_CENTER_LOG("hanlde malloc fail\r\n");
return NULL;
}
handle->exit_flg = 0;
handle->eventSubscriberHead = (EventSubscriberNode_t *)malloc(sizeof(EventSubscriberNode_t));
if(handle->eventSubscriberHead == NULL)
{
MSG_CENTER_LOG("eventSubscriberHead malloc fail\r\n");
goto FAIL;
}
memset(handle->eventSubscriberHead->event, 0, sizeof(handle->eventSubscriberHead->event));
handle->eventSubscriberHead->func = NULL;
handle->eventSubscriberHead->next = NULL;
handle->eventPublishHead = (EventPublishNode_t *)malloc(sizeof(EventPublishNode_t));
if(handle->eventPublishHead == NULL)
{
MSG_CENTER_LOG("eventPublishHead malloc fail\r\n");
goto FAIL;
}
memset(handle->eventPublishHead->event, 0, sizeof(handle->eventPublishHead->event));
handle->eventPublishHead->next = NULL;
handle->eventPublishHead->data = NULL;
handle->mutexSubscriber = mutexSubscriber;
handle->mutexPublish = mutexPublish;
return handle;
FAIL:
if(handle->eventSubscriberHead)
{
free(handle->eventSubscriberHead);
handle->eventSubscriberHead = NULL;
}
if(handle->eventPublishHead)
{
free(handle->eventPublishHead);
handle->eventPublishHead = NULL;
}
return NULL;
}
void MsgCenterDeinit(MsgCenterHandle_t **msg_handle)
{
MsgCenterHandle_t *handle = *msg_handle;
if(handle == NULL)
{
MSG_CENTER_LOG("handle is null\r\n");
return;
}
MsgMutexLock(handle->mutexSubscriber, true);
EventSubscriberNode_t *subCur = handle->eventSubscriberHead;
EventSubscriberNode_t *subNext = subCur->next;
while(subCur)
{
free(subCur);
subCur = subNext;
if(subCur)
subNext = subCur->next;
}
handle->eventSubscriberHead = NULL;
MsgMutexLock(handle->mutexSubscriber, false);
MsgMutexLock(handle->mutexPublish, true);
EventPublishNode_t *pubCur = handle->eventPublishHead;
EventPublishNode_t *pubNext = pubCur->next;
while(pubCur)
{
if(pubCur->data)
{
free(pubCur->data);
}
free(pubCur);
pubCur = pubNext;
if(pubCur)
pubNext = pubCur->next;
}
handle->eventPublishHead = NULL;
MsgMutexLock(handle->mutexPublish, false);
handle->exit_flg = 1;
free(*msg_handle);
*msg_handle = NULL;
return;
}
int SubscriberEvent(MsgCenterHandle_t *handle, char * event, HandleEvent func, int needId)
{
if(handle == NULL || func == NULL || event == NULL)
{
MSG_CENTER_LOG("handle|func|event is null\r\n");
return -1;
}
EventSubscriberNode_t *node = (EventSubscriberNode_t *)malloc(sizeof(EventSubscriberNode_t));
if(node == NULL)
{
MSG_CENTER_LOG("node is malloc fail\r\n");
return -1;
}
node->func = func;
node->needId = needId;
strncpy(node->event, event, sizeof(node->event) - 1);
node->next = NULL;
MsgMutexLock(handle->mutexSubscriber, true);
EventSubscriberNode_t *current = handle->eventSubscriberHead;
if(current != NULL)
{
while (current->next != NULL)
{
current = current->next;
if(strcmp(current->event, event) == 0 && current->func == func && current->needId == needId)
{
MSG_CENTER_LOG("exist same node %s\r\n", event);
free(node);
MsgMutexLock(handle->mutexSubscriber, false);
return -1;
}
}
current->next = node;
}
else
{
free(node);
MsgMutexLock(handle->mutexSubscriber, false);
return -1;
}
MsgMutexLock(handle->mutexSubscriber, false);
return 0;
}
int ReleaseSubscriberEvent(MsgCenterHandle_t *handle, char *event, HandleEvent func)
{
if(handle == NULL || event == NULL || func == NULL)
{
MSG_CENTER_LOG("handle|func|event is null\r\n");
return -1;
}
MsgMutexLock(handle->mutexSubscriber, true);
EventSubscriberNode_t **current = &handle->eventSubscriberHead;
while ((*current) != NULL)
{
if(strcmp((*current)->event, event) == 0 && (*current)->func == func)
{
if((*current)->next == NULL)
{
(*current)->func = NULL;
free(*current);
(*current) = NULL;
break;
}
EventSubscriberNode_t *de = *current;
*current = (*current)->next;
free(de);
break;
}
current = &(*current)->next;
}
MsgMutexLock(handle->mutexSubscriber, false);
return 0;
}
int PublishEvent(MsgCenterHandle_t *handle, char * event, void *data, int datalen, int srcId)
{
if(handle == NULL || event == NULL)
{
MSG_CENTER_LOG("handle|event is null\r\n");
return -1;
}
EventPublishNode_t *node = (EventPublishNode_t *)malloc(sizeof(EventPublishNode_t));
if(node == NULL)
{
MSG_CENTER_LOG("node is malloc fail\r\n");
return -1;
}
node->srcId = srcId;
strncpy(node->event, event, sizeof(node->event) - 1);
node->data = malloc(datalen);
if(node->data == NULL)
{
MSG_CENTER_LOG("node->data is malloc fail\r\n");
free(node);
return -1;
}
memcpy(node->data, data, datalen);
node->next = NULL;
MsgMutexLock(handle->mutexPublish, true);
EventPublishNode_t *current = handle->eventPublishHead;
if(current != NULL)
{
while (current->next != NULL)
{
current = current->next;
}
current->next = node;
}
else
{
free(node);
MsgMutexLock(handle->mutexPublish, false);
return -1;
}
MsgMutexLock(handle->mutexPublish, false);
return 0;
}