juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java

   1 package de.juplo.kafka.wordcount.splitter;
   2
   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.context.annotation.ImportResource;
   8 import org.springframework.expression.common.LiteralExpression;
   9 import org.springframework.integration.annotation.InboundChannelAdapter;
  10 import org.springframework.integration.annotation.ServiceActivator;
  11 import org.springframework.integration.config.EnableIntegration;
  12 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
  13 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
  14 import org.springframework.integration.support.MessageBuilder;
  15 import org.springframework.kafka.core.ConsumerFactory;
  16 import org.springframework.kafka.core.KafkaTemplate;
  17 import org.springframework.kafka.listener.ConsumerProperties;
  18 import org.springframework.kafka.support.KafkaHeaders;
  19 import org.springframework.messaging.Message;
  20 import org.springframework.messaging.MessageChannel;
  21 import org.springframework.messaging.MessageHandler;
  22 import org.springframework.messaging.MessageHeaders;
  23 import org.springframework.messaging.support.ChannelInterceptor;
  24
  25
  26 @SpringBootApplication
  27 @EnableConfigurationProperties(SplitterApplicationProperties.class)
  28 @EnableIntegration
  29 @ImportResource("classpath:integration.xml")
  30 public class SplitterApplication
  31 {
  32         @Bean
  33         ChannelInterceptor keyInterceptor ()
  34         {
  35                 return new ChannelInterceptor()
  36                 {
  37                         @Override
  38                         public Message<?> preSend(Message<?> message, MessageChannel channel)
  39                         {
  40                                 MessageHeaders headers = message.getHeaders();
  41                                 Object key = headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY);
  42                                 return
  43                                                 MessageBuilder
  44                                                                 .fromMessage(message)
  45                                                                 .setHeader(KafkaHeaders.MESSAGE_KEY, key)
  46                                                                 .build();
  47                         }
  48                 };
  49         }
  50
  51         @InboundChannelAdapter(channel = "recordings")
  52         @Bean
  53         KafkaMessageSource<String, String> source(
  54                         ConsumerFactory<String, String> cf,
  55                         SplitterApplicationProperties properties)
  56         {
  57                 return new KafkaMessageSource<>(cf, new ConsumerProperties(properties.getInputTopic()));
  58         }
  59
  60         @Bean
  61         @ServiceActivator(inputChannel = "words")
  62         MessageHandler handler(
  63                         KafkaTemplate<String, String> kafkaTemplate,
  64                         SplitterApplicationProperties properties)
  65         {
  66                 KafkaProducerMessageHandler<String, String> handler =
  67                                 new KafkaProducerMessageHandler<>(kafkaTemplate);
  68                 handler.setTopicExpression(new LiteralExpression(properties.getOutputTopic()));
  69                 return handler;
  70         }
  71
  72         public static void main(String[] args)
  73         {
  74                 SpringApplication.run(SplitterApplication.class, args);
  75         }
  76 }