X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationConfiguration.java;h=f83661eec06334f10133cd82ca1302f5eaeb1b1b;hb=a4be0eac8d9f234cd53ae1917a084f65d91d0460;hp=973e973271d601ff4b0d8026d413505c0bf6f4ce;hpb=c9d7601fc551069cd3a77da06a6333f22101a8a0;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index 973e973..f83661e 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -18,21 +18,29 @@ import java.util.concurrent.Executors; public class ApplicationConfiguration { @Bean - public AdderRecordHandler sumRecordHandler() + public ApplicationRecordHandler recordHandler(AdderResults adderResults) { - return new AdderRecordHandler(); + return new ApplicationRecordHandler(adderResults); } @Bean - public AdderRebalanceListener sumRebalanceListener( - AdderRecordHandler adderRecordHandler, - PartitionStatisticsRepository repository, + public AdderResults adderResults() + { + return new AdderResults(); + } + + @Bean + public ApplicationRebalanceListener rebalanceListener( + ApplicationRecordHandler recordHandler, + AdderResults adderResults, + StateRepository stateRepository, Consumer consumer, ApplicationProperties properties) { - return new AdderRebalanceListener( - adderRecordHandler, - repository, + return new ApplicationRebalanceListener( + recordHandler, + adderResults, + stateRepository, properties.getClientId(), properties.getTopic(), Clock.systemDefaultZone(), @@ -44,8 +52,8 @@ public class ApplicationConfiguration public EndlessConsumer endlessConsumer( KafkaConsumer kafkaConsumer, ExecutorService executor, - AdderRebalanceListener adderRebalanceListener, - AdderRecordHandler adderRecordHandler, + ApplicationRebalanceListener rebalanceListener, + ApplicationRecordHandler recordHandler, ApplicationProperties properties) { return @@ -54,8 +62,8 @@ public class ApplicationConfiguration properties.getClientId(), properties.getTopic(), kafkaConsumer, - adderRebalanceListener, - adderRecordHandler); + rebalanceListener, + recordHandler); } @Bean