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 3d4b48f..22576a0 100644 (file)
--- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
+++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
@@ -5,19 +5,11 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ImportResource;
-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.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.ConsumerProperties;
-import org.springframework.messaging.MessageHandler;
 
 
 @SpringBootApplication
@@ -35,21 +27,6 @@ public class SplitterApplication
                return new KafkaMessageSource<>(cf, new ConsumerProperties(properties.getInputTopic()));
        }
 
-       @Bean
-       @ServiceActivator(inputChannel = "words")
-       MessageHandler handler(
-                       KafkaTemplate<String, String> kafkaTemplate,
-                       SplitterApplicationProperties properties)
-       {
-               KafkaProducerMessageHandler<String, String> handler =
-                               new KafkaProducerMessageHandler<>(kafkaTemplate);
-               handler.setTopicExpression(new LiteralExpression(properties.getOutputTopic()));
-               final ExpressionParser parser = new SpelExpressionParser();
-               Expression expression = parser.parseExpression("headers['kafka_receivedMessageKey']");
-               handler.setMessageKeyExpression(expression);
-               return handler;
-       }
-
        public static void main(String[] args)
        {
                SpringApplication.run(SplitterApplication.class, args);

diff --git a/src/main/resources/integration.xml b/src/main/resources/integration.xml
index 2495877..4ecea8b 100644 (file)
--- a/src/main/resources/integration.xml
+++ b/src/main/resources/integration.xml
@@ -8,4 +8,11 @@
                http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
         http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
 
+  <int-kafka:inbound-channel-adapter
+      id="recordings"
+      consumer-factory="kafkaConsumerFactory"
+      payload-type="java.lang.String"
+      group-id="splitter"
+      channel="recordings" />
+
 </beans>