xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(stream): fix syntax error.

This commit is contained in:
Haojun Liao 2023-08-25 17:39:31 +08:00
parent 43a3f9f535
commit ef3e2737f4
1 changed files with 8 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
source/libs/stream/src/streamQueue.c
View File

@ -40,6 +40,13 @@ SStreamQueue* streamQueueOpen(int64_t cap) {
return pQueue; return pQueue;
} }
void streamQueueCleanup(SStreamQueue* pQueue) {
void* qItem = NULL;
while ((qItem = streamQueueNextItem(pQueue)) != NULL) {
streamFreeQitem(qItem);
}
pQueue->status = STREAM_QUEUE__SUCESS;
}
// void streamQueueClose(SStreamQueue* pQueue) { // void streamQueueClose(SStreamQueue* pQueue) {
// streamQueueCleanup(pQueue); // streamQueueCleanup(pQueue);
@ -50,12 +57,8 @@ SStreamQueue* streamQueueOpen(int64_t cap) {
void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) { void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {
qDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->queue, taosQueueItemSize(pQueue->queue)); qDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->queue, taosQueueItemSize(pQueue->queue));
void* qItem = NULL; streamQueueCleanup(pQueue);
while ((qItem = streamQueueNextItem(pQueue)) != NULL) {
streamFreeQitem(qItem);
}
pQueue->status = STREAM_QUEUE__SUCESS;
taosFreeQall(pQueue->qall); taosFreeQall(pQueue->qall);
taosCloseQueue(pQueue->queue); taosCloseQueue(pQueue->queue);
taosMemoryFree(pQueue); taosMemoryFree(pQueue);