feat: query redirect

include/common/tmsgdef.h

@@ -208,6 +208,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_SCH_CANCEL_TASK, "cancel-task", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SCH_DROP_TASK, "drop-task", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SCH_EXPLAIN, "explain", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "link-broken", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_STREAM_MSG)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DEPLOY, "stream-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
@@ -217,9 +218,6 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RECOVER, "stream-task-recover", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE, "stream-retrieve", NULL, NULL)
 
-  TD_NEW_MSG_SEG(TDMT_SCH_MSG)
-  TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "link-broken", NULL, NULL)
-
   TD_NEW_MSG_SEG(TDMT_MON_MSG)
   TD_DEF_MSG_TYPE(TDMT_MON_MM_INFO, "monitor-minfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MON_VM_INFO, "monitor-vinfo", NULL, NULL)

source/libs/transport/src/transSvr.c

@@ -1042,7 +1042,7 @@ void transReleaseSrvHandle(void* handle) {
   m->type = Release;
 
   tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;
 _return1:
@@ -1071,7 +1071,7 @@ void transSendResponse(const STransMsg* msg) {
 
   STraceId* trace = (STraceId*)&msg->info.traceId;
   tGTrace("conn %p start to send resp (1/2)", exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;
 _return1:
@@ -1100,7 +1100,7 @@ void transRegisterMsg(const STransMsg* msg) {
   m->type = Register;
 
   tTrace("%s conn %p start to register brokenlink callback", transLabel(pThrd->pTransInst), exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;