1 package de.juplo.kafka.wordcount.splitter;
3 import org.springframework.boot.SpringApplication;
4 import org.springframework.boot.autoconfigure.SpringBootApplication;
5 import org.springframework.boot.context.properties.EnableConfigurationProperties;
6 import org.springframework.context.annotation.Bean;
7 import org.springframework.expression.Expression;
8 import org.springframework.expression.ExpressionParser;
9 import org.springframework.expression.common.LiteralExpression;
10 import org.springframework.expression.spel.standard.SpelExpressionParser;
11 import org.springframework.integration.annotation.ServiceActivator;
12 import org.springframework.integration.config.EnableIntegration;
13 import org.springframework.integration.core.MessageProducer;
14 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
15 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
16 import org.springframework.kafka.core.ConsumerFactory;
17 import org.springframework.kafka.core.KafkaTemplate;
18 import org.springframework.kafka.listener.ContainerProperties;
19 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
20 import org.springframework.messaging.MessageHandler;
23 @SpringBootApplication
24 @EnableConfigurationProperties(SplitterApplicationProperties.class)
26 public class SplitterApplication
29 MessageProducer messageProducer(
30 KafkaMessageListenerContainer<String, String> messageListenerContainer)
32 KafkaMessageDrivenChannelAdapter<String, String> adapter =
33 new KafkaMessageDrivenChannelAdapter(messageListenerContainer);
34 adapter.setOutputChannelName("recordings");
39 KafkaMessageListenerContainer<String, String> messageListenerContainer(
40 ConsumerFactory<String,String> consumerFactory,
41 SplitterApplicationProperties properties)
44 new KafkaMessageListenerContainer<>(
46 new ContainerProperties(properties.getInputTopic()));
50 @ServiceActivator(inputChannel = "words")
51 MessageHandler handler(
52 KafkaTemplate<String, String> kafkaTemplate,
53 SplitterApplicationProperties properties)
55 KafkaProducerMessageHandler<String, String> handler =
56 new KafkaProducerMessageHandler<>(kafkaTemplate);
57 handler.setTopicExpression(new LiteralExpression(properties.getOutputTopic()));
58 final ExpressionParser parser = new SpelExpressionParser();
59 Expression expression = parser.parseExpression("headers['kafka_receivedMessageKey']");
60 handler.setMessageKeyExpression(expression);
64 public static void main(String[] args)
66 SpringApplication.run(SplitterApplication.class, args);