0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

minor improvements

Browse source
This commit is contained in:
iliax 2023-08-14 01:31:00 +04:00
parent c685bf057a
commit 5aa6352df4
1 changed files with 9 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java
View file

@ -13,12 +13,12 @@ class ConsumingStats {
void sendConsumingEvt(FluxSink<TopicMessageEventDTO> sink, PolledRecords polledRecords) {
bytes += polledRecords.bytes();
this.records += polledRecords.count();
this.elapsed += polledRecords.elapsed().toMillis();
records += polledRecords.count();
elapsed += polledRecords.elapsed().toMillis();
sink.next(
new TopicMessageEventDTO()
.type(TopicMessageEventDTO.TypeEnum.CONSUMING)
.consuming(createConsumingStats(sink))
.consuming(createConsumingStats())
);
}
@ -30,16 +30,16 @@ class ConsumingStats {
sink.next(
new TopicMessageEventDTO()
.type(TopicMessageEventDTO.TypeEnum.DONE)
.consuming(createConsumingStats(sink))
.consuming(createConsumingStats())
);
}
private TopicMessageConsumingDTO createConsumingStats(FluxSink<TopicMessageEventDTO> sink) {
private TopicMessageConsumingDTO createConsumingStats() {
return new TopicMessageConsumingDTO()
.bytesConsumed(this.bytes)
.elapsedMs(this.elapsed)
.isCancelled(sink.isCancelled())
.bytesConsumed(bytes)
.elapsedMs(elapsed)
.isCancelled(false)
.filterApplyErrors(filterApplyErrors)
.messagesConsumed(this.records);
.messagesConsumed(records);
}
}