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

fix(stream): add more check.

This commit is contained in:
Haojun Liao 2023-09-20 16:49:55 +08:00
parent c403a15f36
commit 1ebafe4a1f
1 changed files with 5 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/stream/src/stream.c
View File

@ -241,6 +241,11 @@ int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, S
} }
} }
int8_t st = pTask->status.taskStatus;
if (st == TASK_STATUS__HALT) {
status = TASK_INPUT_STATUS__BLOCKED;
}
{ {
// do send response with the input status // do send response with the input status
int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont); int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont);
@ -255,7 +260,6 @@ int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, S
tDeleteStreamDispatchReq(pReq); tDeleteStreamDispatchReq(pReq);
int8_t st = pTask->status.taskStatus;
if (st == TASK_STATUS__NORMAL || st == TASK_STATUS__SCAN_HISTORY || st == TASK_STATUS__CK) { if (st == TASK_STATUS__NORMAL || st == TASK_STATUS__SCAN_HISTORY || st == TASK_STATUS__CK) {
int8_t schedStatus = streamTaskSetSchedStatusWait(pTask); int8_t schedStatus = streamTaskSetSchedStatusWait(pTask);
if (schedStatus == TASK_SCHED_STATUS__INACTIVE) { if (schedStatus == TASK_SCHED_STATUS__INACTIVE) {