- QueryApplicationProperties properties,
- ObjectMapper mapper,
- ConfigurableApplicationContext context)
+ Properties props,
+ HostInfo applicationServer,
+ String usersInputTopic,
+ String rankingInputTopic,
+ KeyValueBytesStoreSupplier storeSupplier)
+ {
+ Topology topology = buildTopology(
+ usersInputTopic,
+ rankingInputTopic,
+ storeSupplier);
+ streams = new KafkaStreams(topology, props);
+ hostInfo = applicationServer;
+ storeParameters = StoreQueryParameters.fromNameAndType(STORE_NAME, QueryableStoreTypes.keyValueStore());;
+ }
+
+ static Topology buildTopology(
+ String usersInputTopic,
+ String rankingInputTopic,
+ KeyValueBytesStoreSupplier storeSupplier)