8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2026-01-07 23:28:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: 修复 Consume 点击 Stop 后未停止请求发送的问题

Browse Source
This commit is contained in:
GraceWalk
2022-09-13 14:24:30 +08:00
parent 6effba69a0
commit d550fc5068
1 changed files with 25 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
km-console/packages/layout-clusters-fe/src/pages/TestingConsumer/Consume.tsx
View File

@@ -178,38 +178,35 @@ const ConsumeClientTest = () => {
partitionProcessRef.current = processList; partitionProcessRef.current = processList;
curPartitionList.current = _partitionList; curPartitionList.current = _partitionList;
if (!isStopStatus.current) {
switch (until) { switch (until) {
case 'timestamp': case 'timestamp':
setIsStop(currentTime >= untilDate); setIsStop(currentTime >= untilDate);
isStopStatus.current = currentTime >= untilDate; isStopStatus.current = currentTime >= untilDate;
break; break;
case 'number of messages': case 'number of messages':
setIsStop(+recordCountCur.current >= untilMsgNum); setIsStop(+recordCountCur.current >= untilMsgNum);
isStopStatus.current = +recordCountCur.current >= untilMsgNum; isStopStatus.current = +recordCountCur.current >= untilMsgNum;
break; break;
case 'number of messages per partition': // 所有分区都达到了设定值 case 'number of messages per partition': // 所有分区都达到了设定值
// 过滤出消费数量不足设定值的partition // 过滤出消费数量不足设定值的partition
const filtersPartition = _partitionList.filter((item: any) => item.recordCount < untilMsgNum); const filtersPartition = _partitionList.filter((item: any) => item.recordCount < untilMsgNum);
curPartitionList.current = filtersPartition; // 用作下一次请求的入参 curPartitionList.current = filtersPartition; // 用作下一次请求的入参
if (!isStop) {
setIsStop(filtersPartition.length < 1); setIsStop(filtersPartition.length < 1);
isStopStatus.current = filtersPartition.length < 1; isStopStatus.current = filtersPartition.length < 1;
} break;
break; case 'max size':
case 'max size': setIsStop(+recordSizeCur.current >= unitMsgSize);
setIsStop(+recordSizeCur.current >= unitMsgSize); isStopStatus.current = +recordSizeCur.current >= unitMsgSize;
isStopStatus.current = +recordSizeCur.current >= unitMsgSize; break;
break; case 'max size per partition':
case 'max size per partition': // 过滤出消费size不足设定值的partition
// 过滤出消费size不足设定值的partition const filters = partitionConsumedList.filter((item: any) => item.recordSizeUnitB < unitMsgSize);
const filters = partitionConsumedList.filter((item: any) => item.recordSizeUnitB < unitMsgSize);
if (!isStop) {
setIsStop(filters.length < 1); setIsStop(filters.length < 1);
isStopStatus.current = filters.length < 1; isStopStatus.current = filters.length < 1;
} curPartitionList.current = filters;
curPartitionList.current = filters; break;
break; }
} }
}; };