#4788

Committed 14 Oct 2025 11:21AM UTC coverage: 60.992% (-2.3%) from 63.264%

Build # #4788

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge 7ca9b50f9 into 19574fe21

Run Details

154868 of 324306 branches covered (47.75%)

Branch coverage included in aggregate %.

207304 of 269498 relevant lines covered (76.92%)

125773493.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.73

/source/dnode/mnode/impl/src/mndStreamHb.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "mndStream.h"
#include "mndTrans.h"
#include "mndMnode.h"
#include "tmisce.h"


void mndStreamHbSendRsp(SRpcHandleInfo *pRpcInfo, SRpcMsg* pRsp) {
  tmsgSendRsp(pRsp);
  pRpcInfo->handle = NULL;  // disable auto rsp
}

int32_t mndProcessStreamHb(SRpcMsg *pReq) {
  SMnode      *pMnode = pReq->info.node;
  SStreamHbMsg req = {0};
  SMStreamHbRspMsg rsp = {0};
  int32_t      code = 0;
  int32_t      lino = 0;
  SDecoder     decoder = {0};
  char*        msg = POINTER_SHIFT(pReq->pCont, sizeof(SStreamMsgGrpHeader));
  int32_t      len = pReq->contLen - sizeof(SStreamMsgGrpHeader);
  int64_t      currTs = taosGetTimestampMs();
  SRpcMsg      rspMsg = {0};

  mstDebug("start to process stream hb req msg");

  rsp.streamGId = req.streamGId;

  if ((code = grantCheckExpire(TSDB_GRANT_STREAMS)) < 0) {
    TAOS_CHECK_EXIT(msmHandleGrantExpired(pMnode, code));
  }

  tDecoderInit(&decoder, msg, len);
  code = tDecodeStreamHbMsg(&decoder, &req);
  if (code < 0) {
    mstError("failed to decode stream hb msg, error:%s", tstrerror(terrno));
    tCleanupStreamHbMsg(&req, true);
    tDecoderClear(&decoder);
    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);
  }
  tDecoderClear(&decoder);

  mstDebug("start to process grp %d stream-hb from dnode:%d, snodeId:%d, vgLeaders:%d, streamStatus:%d", 
      req.streamGId, req.dnodeId, req.snodeId, (int32_t)taosArrayGetSize(req.pVgLeaders), (int32_t)taosArrayGetSize(req.pStreamStatus));

  (void)msmHandleStreamHbMsg(pMnode, currTs, &req, pReq, &rspMsg);

_exit:

  if (code) {
    msmEncodeStreamHbRsp(code, &pReq->info, &rsp, &rspMsg);
  }
  
  mndStreamHbSendRsp(&pReq->info, &rspMsg);
  tCleanupStreamHbMsg(&req, true);
  tFreeSMStreamHbRspMsg(&rsp);

  mstDebug("end to process stream hb req msg, code:%d", code);

  return code;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "mndStream.h"
17	#include "mndTrans.h"
18	#include "mndMnode.h"
19	#include "tmisce.h"
20
21
22	void mndStreamHbSendRsp(SRpcHandleInfo pRpcInfo, SRpcMsg pRsp) {	13,489,995✔
23	tmsgSendRsp(pRsp);	13,489,995✔
24	pRpcInfo->handle = NULL; // disable auto rsp	13,489,995✔
25	}	13,489,995✔
26
27	int32_t mndProcessStreamHb(SRpcMsg *pReq) {	13,489,995✔
28	SMnode *pMnode = pReq->info.node;	13,489,995✔
29	SStreamHbMsg req = {0};	13,489,995✔
30	SMStreamHbRspMsg rsp = {0};	13,489,995✔
31	int32_t code = 0;	13,489,995✔
32	int32_t lino = 0;	13,489,995✔
33	SDecoder decoder = {0};	13,489,995✔
34	char* msg = POINTER_SHIFT(pReq->pCont, sizeof(SStreamMsgGrpHeader));	13,489,995✔
35	int32_t len = pReq->contLen - sizeof(SStreamMsgGrpHeader);	13,489,995✔
36	int64_t currTs = taosGetTimestampMs();	13,489,995✔
37	SRpcMsg rspMsg = {0};	13,489,995✔
38
39	mstDebug("start to process stream hb req msg");	13,489,995✔
40
41	rsp.streamGId = req.streamGId;	13,489,995✔
42
43	if ((code = grantCheckExpire(TSDB_GRANT_STREAMS)) < 0) {	13,489,995!
44	TAOS_CHECK_EXIT(msmHandleGrantExpired(pMnode, code));	×
45	}
46
47	tDecoderInit(&decoder, msg, len);	13,489,995✔
48	code = tDecodeStreamHbMsg(&decoder, &req);	13,489,995✔
49	if (code < 0) {	13,489,995!
50	mstError("failed to decode stream hb msg, error:%s", tstrerror(terrno));	×
51	tCleanupStreamHbMsg(&req, true);	×
52	tDecoderClear(&decoder);	×
53	TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);	×
54	}
55	tDecoderClear(&decoder);	13,489,995✔
56
57	mstDebug("start to process grp %d stream-hb from dnode:%d, snodeId:%d, vgLeaders:%d, streamStatus:%d",	13,489,995✔
58	req.streamGId, req.dnodeId, req.snodeId, (int32_t)taosArrayGetSize(req.pVgLeaders), (int32_t)taosArrayGetSize(req.pStreamStatus));
59
60	(void)msmHandleStreamHbMsg(pMnode, currTs, &req, pReq, &rspMsg);	13,489,995✔
61
62	_exit:	13,489,995✔
63
64	if (code) {	13,489,995!
65	msmEncodeStreamHbRsp(code, &pReq->info, &rsp, &rspMsg);	×
66	}
67
68	mndStreamHbSendRsp(&pReq->info, &rspMsg);	13,489,995✔
69	tCleanupStreamHbMsg(&req, true);	13,489,995✔
70	tFreeSMStreamHbRspMsg(&rsp);	13,489,995✔
71
72	mstDebug("end to process stream hb req msg, code:%d", code);	13,489,995✔
73
74	return code;	13,489,995✔
75	}

taosdata / TDengine / #4788

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous