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

wip

Browse source
This commit is contained in:
iliax 2023-08-07 16:55:35 +04:00
parent f985f0e360
commit fda75b4a5e
1 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java
View file

@ -18,6 +18,12 @@ public abstract class AbstractEmitter implements java.util.function.Consumer<Flu
this.pollingSettings = pollingSettings;
}
private PolledRecords poll(FluxSink<TopicMessageEventDTO> sink, EnhancedConsumer consumer, Duration timeout) {
var records = consumer.pollEnhanced(timeout);
sendConsuming(sink, records);
return records;
}
protected PolledRecords poll(FluxSink<TopicMessageEventDTO> sink, EnhancedConsumer consumer) {
return poll(sink, consumer, pollingSettings.getPollTimeout());
}
@ -26,12 +32,6 @@ public abstract class AbstractEmitter implements java.util.function.Consumer<Flu
return poll(sink, consumer, pollingSettings.getPartitionPollTimeout());
}
private PolledRecords poll(FluxSink<TopicMessageEventDTO> sink, EnhancedConsumer consumer, Duration timeout) {
var records = consumer.pollEnhanced(timeout);
sendConsuming(sink, records);
return records;
}
protected void buffer(ConsumerRecord<Bytes, Bytes> rec) {
messagesProcessing.buffer(rec);
}