gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ feat(add publish at once API): define in mq_internal.h temporarily

Browse Source
This commit is contained in:
yangwei
2024-09-25 18:36:31 +08:00
parent 74f77f3411
commit 75d60bb3ea
2 changed files with 39 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
infra/mq/mq.c
View File

@@ -106,25 +106,46 @@ int mq_schema_destroy_topic(struct mq_schema *s, int topic_id)
} }
static void mq_dispatch_one_message(struct mq_schema *s, struct mq_message *mq_elt) static int mq_dispatch_one_message(struct mq_topic *topic, struct mq_message *mq_elt)
{ {
struct mq_subscriber *sub_elt, *sub_tmp; struct mq_subscriber *sub_elt, *sub_tmp;
struct mq_topic *topic = (struct mq_topic *)utarray_eltptr(s->topic_array,(unsigned int)(mq_elt->header.topic_id)); if(topic==NULL || mq_elt==NULL)return -1;
if (topic) DL_FOREACH_SAFE(topic->subscribers, sub_elt, sub_tmp)
{ {
DL_FOREACH_SAFE(topic->subscribers, sub_elt, sub_tmp) if (sub_elt->msg_cb)
{ {
if (sub_elt->msg_cb) if (topic->dispatch_cb)
{ topic->dispatch_cb(mq_elt->header.topic_id, mq_elt->body, sub_elt->msg_cb, sub_elt->msg_cb_arg,
if(topic->dispatch_cb)topic->dispatch_cb(mq_elt->header.topic_id, topic->dispatch_cb_arg);
mq_elt->body, else
sub_elt->msg_cb, sub_elt->msg_cb(mq_elt->header.topic_id, mq_elt->body, sub_elt->msg_cb_arg);
sub_elt->msg_cb_arg,
topic->dispatch_cb_arg);
else sub_elt->msg_cb(mq_elt->header.topic_id, mq_elt->body, sub_elt->msg_cb_arg);
}
} }
} }
return 0;
}
int mq_runtime_publish_message_at_once(struct mq_runtime *rt, int topic_id, void *msg)
{
if(rt==NULL || rt->schema == NULL || rt->schema->topic_array == NULL)return -1;
//if(rt->is_cleaning==true)return -1;
unsigned int len = utarray_len(rt->schema->topic_array);
if (len <= (unsigned int)topic_id)return -1;
struct mq_topic *topic = (struct mq_topic *)utarray_eltptr(rt->schema->topic_array,(unsigned int)(topic_id));
if(topic==NULL)return -1;
struct mq_message *mq_elt = CALLOC(struct mq_message,1);
mq_elt->rt=rt;
mq_elt->header.topic_id = topic_id;
mq_elt->header.priority = STELLAR_MQ_PRIORITY_HIGH;
mq_elt->body = msg;
mq_dispatch_one_message(topic, mq_elt);
if (topic->free_cb)
{
topic->free_cb(mq_elt->body, topic->free_cb_arg);
}
FREE(mq_elt);
return 0;
} }
void mq_runtime_clean(struct mq_runtime *rt) void mq_runtime_clean(struct mq_runtime *rt)
@@ -155,6 +176,7 @@ void mq_runtime_clean(struct mq_runtime *rt)
void mq_runtime_dispatch(struct mq_runtime *rt) void mq_runtime_dispatch(struct mq_runtime *rt)
{ {
struct mq_topic *topic=NULL;
struct mq_message *mq_elt=NULL, *mq_tmp=NULL; struct mq_message *mq_elt=NULL, *mq_tmp=NULL;
int cur_priority = STELLAR_MQ_PRIORITY_HIGH; int cur_priority = STELLAR_MQ_PRIORITY_HIGH;
while(cur_priority >= STELLAR_MQ_PRIORITY_LOW) while(cur_priority >= STELLAR_MQ_PRIORITY_LOW)
@@ -168,7 +190,8 @@ void mq_runtime_dispatch(struct mq_runtime *rt)
{ {
DL_DELETE(rt->priority_mq[mq_elt->header.priority], mq_elt); DL_DELETE(rt->priority_mq[mq_elt->header.priority], mq_elt);
rt->priority_mq_len[mq_elt->header.priority]-=1; rt->priority_mq_len[mq_elt->header.priority]-=1;
mq_dispatch_one_message(rt->schema, mq_elt); topic = (struct mq_topic *)utarray_eltptr(rt->schema->topic_array,(unsigned int)(mq_elt->header.topic_id));
mq_dispatch_one_message(topic, mq_elt);
DL_APPEND(rt->priority_mq[STELLAR_MQ_DEATH_LETTER], mq_elt); // move to dlq list DL_APPEND(rt->priority_mq[STELLAR_MQ_DEATH_LETTER], mq_elt); // move to dlq list
rt->priority_mq_len[STELLAR_MQ_DEATH_LETTER]+=1; rt->priority_mq_len[STELLAR_MQ_DEATH_LETTER]+=1;
cur_priority=STELLAR_MQ_PRIORITY_HIGH; cur_priority=STELLAR_MQ_PRIORITY_HIGH;

3
infra/mq/mq_internal.h
View File

@@ -70,7 +70,8 @@ struct mq_runtime
bool is_cleaning; bool is_cleaning;
}; };
int mq_runtime_publish_message_with_priority(struct mq_runtime *rt, int topic_id, void *data, enum mq_property priority); int mq_runtime_publish_message_with_priority(struct mq_runtime *rt, int topic_id, void *msg, enum mq_property priority);
int mq_runtime_publish_message_at_once(struct mq_runtime *rt, int topic_id, void *msg);
#ifdef __cplusplus #ifdef __cplusplus
} }