X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fquery%2FQueryApplication.java;h=eeee7eba22b5e0a14aefe21af931570aa659105b;hb=57f47fca712981116b726e437f589ac727c5d0a7;hp=813d3b28212acc5b059c2ae4df41b8f45fc535fa;hpb=8b1fd41f91be906ac0c691b4df7c528cad9ff583;p=demos%2Fkafka%2Fwordcount diff --git a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplication.java b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplication.java index 813d3b2..eeee7eb 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplication.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplication.java @@ -1,56 +1,12 @@ package de.juplo.kafka.wordcount.query; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.kafka.streams.state.HostInfo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.autoconfigure.web.ServerProperties; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; - -import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.Socket; @SpringBootApplication -@EnableConfigurationProperties(QueryApplicationProperties.class) public class QueryApplication { - @Bean - public QueryStreamProcessor usersStreamProcessor( - ServerProperties serverProperties, - QueryApplicationProperties properties, - ObjectMapper mapper, - ConfigurableApplicationContext context) throws IOException - { - String host; - if (serverProperties.getAddress() == null) - { - HostInfo bootstrapServer = HostInfo.buildFromEndpoint(properties.getBootstrapServer()); - Socket socket = new Socket(); - socket.connect(new InetSocketAddress(bootstrapServer.host(), bootstrapServer.port())); - host = socket.getLocalAddress().getHostAddress(); - } - else - { - host = serverProperties.getAddress().getHostAddress(); - } - - Integer port = serverProperties.getPort() == null ? 8080 : serverProperties.getPort(); - - return new QueryStreamProcessor( - properties.getApplicationId(), - new HostInfo(host, port), - properties.getBootstrapServer(), - properties.getUsersInputTopic(), - properties.getRankingInputTopic(), - mapper, - context); - } - - public static void main(String[] args) { SpringApplication.run(QueryApplication.class, args);