public class KafkaServicesConfiguration
{
@Bean
- ChannelTaskRunner channelTaskRunner(
- ChannelTaskExecutor infoChannelTaskExecutor,
- ChannelTaskExecutor dataChannelTaskExecutor)
+ KafkaServicesThreadPoolTaskExecutorCustomizer kafkaServicesThreadPoolTaskExecutorCustomizer()
{
- return new ChannelTaskRunner(
- infoChannelTaskExecutor,
- dataChannelTaskExecutor);
+ return new KafkaServicesThreadPoolTaskExecutorCustomizer();
}
- @Bean
+ @Bean(initMethod = "executeChannelTask", destroyMethod = "join")
ChannelTaskExecutor infoChannelTaskExecutor(
ThreadPoolTaskExecutor taskExecutor,
InfoChannel infoChannel,
};
}
- @Bean
+ @Bean(initMethod = "executeChannelTask", destroyMethod = "join")
ChannelTaskExecutor dataChannelTaskExecutor(
ThreadPoolTaskExecutor taskExecutor,
DataChannel dataChannel,
zoneId,
properties.getKafka().getNumPartitions(),
properties.getKafka().getPollingInterval(),
- properties.getChatroomBufferSize(),
+ properties.getChatroomHistoryLimit(),
clock,
channelMediator,
shardingPublisherStrategy);
{
return ZoneId.systemDefault();
}
+
+ @Bean
+ ChannelReactiveHealthIndicator dataChannelHealthIndicator(
+ DataChannel dataChannel)
+ {
+ return new ChannelReactiveHealthIndicator(dataChannel);
+ }
+
+ @Bean
+ ChannelReactiveHealthIndicator infoChannelHealthIndicator(InfoChannel infoChannel)
+ {
+ return new ChannelReactiveHealthIndicator(infoChannel);
+ }
}