- List<String> topics = List.of(properties.getKafka().getChatRoomChannelTopic());
- chatRoomChannelConsumer.subscribe(topics, chatRoomChannel);
- log.info("Starting the consumer for the ChatRoomChannel");
- chatRoomChannelConsumerJob = taskExecutor
- .submitCompletable(chatRoomChannel)
- .exceptionally(e ->
- {
- log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
- return null;
- });
- }
-
- @PreDestroy
- public void joinChatRoomChannelConsumerJob()
- {
- log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
- chatRoomChannelConsumer.wakeup();
- log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
- chatRoomChannelConsumerJob.join();
- log.info("Joined the consumer of the ChatRoomChannel");