From 0bcf271cffe6d7263d76dc6618af9a03b4cb6179 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 17 Jul 2022 19:03:18 +0200
Subject: [PATCH] splitter: 1.0.0-spring-ingetration - message-driven
 channel-adapter

* Instead of the `@InboundChannelAdapter`-annotation, a
  `KafkaMessageDrivenChannelAdapter` can be instanciated.
* The adapter takes a `AbstractMessageListenerContainer` as constructor
  argument.
* The output-channel must be set explicitly on the created instance.
* This approach is advantageous, if technical details are important,
  because the used `MessageListenerContainer` can be configured explicitly.
---
 .../splitter/SplitterApplication.java         | 28 +++++++++++++------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
index 246fffd..b5948c3 100644
--- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
+++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
@@ -8,14 +8,15 @@ import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.listener.ConsumerProperties;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.messaging.MessageHandler;
 
 
@@ -24,15 +25,26 @@ import org.springframework.messaging.MessageHandler;
 @EnableIntegration
 public class SplitterApplication
 {
-	@InboundChannelAdapter(channel = "recordings")
 	@Bean
-	KafkaMessageSource<String, String> source(
-			ConsumerFactory<String, String> cf,
-			SplitterApplicationProperties properties)
+	MessageProducer messageProducer(
+			KafkaMessageListenerContainer<String, String> messageListenerContainer)
 	{
-		return new KafkaMessageSource<>(cf, new ConsumerProperties(properties.getInputTopic()));
+		KafkaMessageDrivenChannelAdapter<String, String> adapter =
+				new KafkaMessageDrivenChannelAdapter(messageListenerContainer);
+		adapter.setOutputChannelName("recordings");
+		return adapter;
 	}
 
+	@Bean
+	KafkaMessageListenerContainer<String, String> messageListenerContainer(
+			ConsumerFactory<String,String> consumerFactory,
+			SplitterApplicationProperties properties)
+	{
+		return
+				new KafkaMessageListenerContainer<>(
+						consumerFactory,
+						new ContainerProperties(properties.getInputTopic()));
+	}
 	@Bean
 	@ServiceActivator(inputChannel = "words")
 	MessageHandler handler(
-- 
2.20.1