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.expression.common.LiteralExpression;
   8 import org.springframework.integration.annotation.InboundChannelAdapter;
   9 import org.springframework.integration.annotation.ServiceActivator;
  10 import org.springframework.integration.channel.DirectChannel;
  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 public class SplitterApplication
  30 {
  31         @Bean
  32         MessageChannel words()
  33         {
  34                 DirectChannel words = new DirectChannel();
  35                 words.addInterceptor(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                 return words;
  50         }
  51
  52         @InboundChannelAdapter(channel = "recordings")
  53         @Bean
  54         KafkaMessageSource<String, String> source(
  55                         ConsumerFactory<String, String> cf,
  56                         SplitterApplicationProperties properties)
  57         {
  58                 return new KafkaMessageSource<>(cf, new ConsumerProperties(properties.getInputTopic()));
  59         }
  60
  61         @Bean
  62         @ServiceActivator(inputChannel = "words")
  63         MessageHandler handler(
  64                         KafkaTemplate<String, String> kafkaTemplate,
  65                         SplitterApplicationProperties properties)
  66         {
  67                 KafkaProducerMessageHandler<String, String> handler =
  68                                 new KafkaProducerMessageHandler<>(kafkaTemplate);
  69                 handler.setTopicExpression(new LiteralExpression(properties.getOutputTopic()));
  70                 return handler;
  71         }
  72
  73         public static void main(String[] args)
  74         {
  75                 SpringApplication.run(SplitterApplication.class, args);
  76         }
  77 }