X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fimplementation%2Fkafka%2FKafkaServicesConfiguration.java;h=58e111734d40b8c20a7bd39ff710951ece8a4353;hb=89b5eb1eaf7890c830848dbc4281add2bff41131;hp=6b7c15662c33992f8a36b78dedcfea178a3adc4a;hpb=e73b46955f6b7aff3aa1ecb79b2aee75fa76b097;p=demos%2Fkafka%2Fchat diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java index 6b7c1566..58e11173 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java @@ -1,7 +1,6 @@ package de.juplo.kafka.chat.backend.implementation.kafka; import de.juplo.kafka.chat.backend.ChatBackendProperties; -import de.juplo.kafka.chat.backend.domain.ChatHomeService; import de.juplo.kafka.chat.backend.domain.ShardingPublisherStrategy; import de.juplo.kafka.chat.backend.implementation.haproxy.HaproxyShardingPublisherStrategy; import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessageTo; @@ -40,25 +39,19 @@ import java.util.Properties; public class KafkaServicesConfiguration { @Bean - ConsumerTaskRunner consumerTaskRunner( - ConsumerTaskExecutor infoChannelConsumerTaskExecutor, - ConsumerTaskExecutor dataChannelConsumerTaskExecutor, - InfoChannel infoChannel) + KafkaServicesThreadPoolTaskExecutorCustomizer kafkaServicesThreadPoolTaskExecutorCustomizer() { - return new ConsumerTaskRunner( - infoChannelConsumerTaskExecutor, - dataChannelConsumerTaskExecutor, - infoChannel); + return new KafkaServicesThreadPoolTaskExecutorCustomizer(); } - @Bean - ConsumerTaskExecutor infoChannelConsumerTaskExecutor( + @Bean(initMethod = "executeChannelTask", destroyMethod = "join") + ChannelTaskExecutor infoChannelTaskExecutor( ThreadPoolTaskExecutor taskExecutor, InfoChannel infoChannel, Consumer infoChannelConsumer, WorkAssignor infoChannelWorkAssignor) { - return new ConsumerTaskExecutor( + return new ChannelTaskExecutor( taskExecutor, infoChannel, infoChannelConsumer, @@ -81,14 +74,14 @@ public class KafkaServicesConfiguration }; } - @Bean - ConsumerTaskExecutor dataChannelConsumerTaskExecutor( + @Bean(initMethod = "executeChannelTask", destroyMethod = "join") + ChannelTaskExecutor dataChannelTaskExecutor( ThreadPoolTaskExecutor taskExecutor, DataChannel dataChannel, Consumer dataChannelConsumer, WorkAssignor dataChannelWorkAssignor) { - return new ConsumerTaskExecutor( + return new ChannelTaskExecutor( taskExecutor, dataChannel, dataChannelConsumer, @@ -124,15 +117,19 @@ public class KafkaServicesConfiguration InfoChannel infoChannel( ChatBackendProperties properties, Producer producer, - Consumer infoChannelConsumer) + Consumer infoChannelConsumer, + ChannelMediator channelMediator) { - return new InfoChannel( + InfoChannel infoChannel = new InfoChannel( properties.getKafka().getInfoChannelTopic(), producer, infoChannelConsumer, properties.getKafka().getPollingInterval(), properties.getKafka().getNumPartitions(), - properties.getKafka().getInstanceUri()); + properties.getKafka().getInstanceUri(), + channelMediator); + channelMediator.setInfoChannel(infoChannel); + return infoChannel; } @Bean @@ -142,10 +139,10 @@ public class KafkaServicesConfiguration Consumer dataChannelConsumer, ZoneId zoneId, Clock clock, - InfoChannel infoChannel, + ChannelMediator channelMediator, ShardingPublisherStrategy shardingPublisherStrategy) { - return new DataChannel( + DataChannel dataChannel = new DataChannel( properties.getInstanceId(), properties.getKafka().getDataChannelTopic(), producer, @@ -153,10 +150,18 @@ public class KafkaServicesConfiguration zoneId, properties.getKafka().getNumPartitions(), properties.getKafka().getPollingInterval(), - properties.getChatroomBufferSize(), + properties.getChatroomHistoryLimit(), clock, - infoChannel, + channelMediator, shardingPublisherStrategy); + channelMediator.setDataChannel(dataChannel); + return dataChannel; + } + + @Bean + ChannelMediator channelMediator() + { + return new ChannelMediator(); } @Bean @@ -305,4 +310,17 @@ public class KafkaServicesConfiguration { return ZoneId.systemDefault(); } + + @Bean + ChannelReactiveHealthIndicator dataChannelHealthIndicator( + DataChannel dataChannel) + { + return new ChannelReactiveHealthIndicator(dataChannel); + } + + @Bean + ChannelReactiveHealthIndicator infoChannelHealthIndicator(InfoChannel infoChannel) + { + return new ChannelReactiveHealthIndicator(infoChannel); + } }