From: Kai Moritz <kai@juplo.de>
Date: Thu, 30 Jun 2022 19:26:25 +0000 (+0200)
Subject: WIP
X-Git-Tag: wip-integration-xml~1
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=8273e88013d9297b3830b50a3a6ce148a51edf76;p=demos%2Fkafka%2Fwordcount

WIP
---

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 5d399f4..e73d2bb 100644
--- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
+++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
@@ -6,12 +6,16 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.ConsumerProperties;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -53,6 +57,15 @@ public class SplitterApplication
 		};
 	}
 
+	@InboundChannelAdapter(channel = "recordings")
+	@Bean
+	KafkaMessageSource<String, String> source(
+			ConsumerFactory<String, String> cf,
+			SplitterApplicationProperties properties)
+	{
+		return new KafkaMessageSource<>(cf, new ConsumerProperties(properties.getInputTopic()));
+	}
+
 	@Bean
 	@ServiceActivator(inputChannel = "words")
 	MessageHandler handler(