X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fquery%2FQueryApplicationConfiguration.java;h=3f866f01090491547b22f7f6b0f229afc5928004;hb=861c055cbdc7fe36fb924c5b694537df8eb06423;hp=813d3b28212acc5b059c2ae4df41b8f45fc535fa;hpb=8681e52b8be6c2970bb229888da6e92260bcef47;p=demos%2Fkafka%2Fwordcount diff --git a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java index 813d3b2..3f866f0 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java @@ -2,33 +2,30 @@ 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 org.springframework.context.annotation.Configuration; import java.io.IOException; import java.net.InetSocketAddress; import java.net.Socket; -@SpringBootApplication +@Configuration @EnableConfigurationProperties(QueryApplicationProperties.class) -public class QueryApplication +public class QueryApplicationConfiguration { @Bean - public QueryStreamProcessor usersStreamProcessor( + public HostInfo applicationServer( ServerProperties serverProperties, - QueryApplicationProperties properties, - ObjectMapper mapper, - ConfigurableApplicationContext context) throws IOException + QueryApplicationProperties applicationProperties) throws IOException { String host; if (serverProperties.getAddress() == null) { - HostInfo bootstrapServer = HostInfo.buildFromEndpoint(properties.getBootstrapServer()); + HostInfo bootstrapServer = HostInfo.buildFromEndpoint(applicationProperties.getBootstrapServer()); Socket socket = new Socket(); socket.connect(new InetSocketAddress(bootstrapServer.host(), bootstrapServer.port())); host = socket.getLocalAddress().getHostAddress(); @@ -40,19 +37,23 @@ public class QueryApplication 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); + return new HostInfo(host, port); } - - public static void main(String[] args) + @Bean + public QueryStreamProcessor streamProcessor( + QueryApplicationProperties applicationProperties, + HostInfo applicationServer, + ObjectMapper mapper, + ConfigurableApplicationContext context) { - SpringApplication.run(QueryApplication.class, args); + return new QueryStreamProcessor( + applicationProperties.getApplicationId(), + applicationServer, + applicationProperties.getBootstrapServer(), + applicationProperties.getUsersInputTopic(), + applicationProperties.getRankingInputTopic(), + mapper, + context); } }