X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fusers%2FUsersStreamProcessor.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fusers%2FUsersStreamProcessor.java;h=ef39e89076cab6b0ccfaf146c75c1ef56161cd54;hb=0f2147d109ce238982953e2d07d9202e3070a67c;hp=0000000000000000000000000000000000000000;hpb=fc7b9dfe12a1401e7365d85ec723364750bdcd3d;p=demos%2Fkafka%2Fwordcount

diff --git a/src/main/java/de/juplo/kafka/wordcount/users/UsersStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/users/UsersStreamProcessor.java
new file mode 100644
index 0000000..ef39e89
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/wordcount/users/UsersStreamProcessor.java
@@ -0,0 +1,118 @@
+package de.juplo.kafka.wordcount.users;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.state.HostInfo;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.springframework.boot.SpringApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.net.URI;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+
+
+@Slf4j
+@Component
+public class UsersStreamProcessor
+{
+	public final KafkaStreams streams;
+	public final HostInfo hostInfo;
+	public final String storeName = "rankingsByUsername";
+	public final StoreQueryParameters<ReadOnlyKeyValueStore<String, String>> storeParameters;
+	public final ObjectMapper mapper;
+
+
+	public UsersStreamProcessor(
+			UsersApplicationProperties properties,
+			ObjectMapper mapper,
+			ConfigurableApplicationContext context)
+	{
+		StreamsBuilder builder = new StreamsBuilder();
+		builder.table(properties.getTopic(), Materialized.as(storeName));
+
+		Properties props = new Properties();
+		props.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId());
+		props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, properties.getApplicationServer());
+		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
+		props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+		props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+		streams = new KafkaStreams(builder.build(), props);
+		streams.setUncaughtExceptionHandler((Throwable e) ->
+		{
+			log.error("Unexpected error!", e);
+			CompletableFuture.runAsync(() ->
+			{
+				log.info("Stopping application...");
+				SpringApplication.exit(context, () -> 1);
+			});
+			return SHUTDOWN_CLIENT;
+		});
+
+		hostInfo = HostInfo.buildFromEndpoint(properties.getApplicationServer());
+		storeParameters = StoreQueryParameters.fromNameAndType(storeName, QueryableStoreTypes.keyValueStore());;
+		this.mapper = mapper;
+	}
+
+	public Optional<URI> getRedirect(String username)
+	{
+		KeyQueryMetadata metadata = streams.queryMetadataForKey(storeName, username, Serdes.String().serializer());
+		HostInfo activeHost = metadata.activeHost();
+		log.debug("Local store for {}: {}, {}:{}", username, metadata.partition(), activeHost.host(), activeHost.port());
+
+		if (activeHost.equals(this.hostInfo) || activeHost.equals(HostInfo.unavailable()))
+		{
+			return Optional.empty();
+		}
+
+		URI location = URI.create("http://" + activeHost.host() + ":" + activeHost.port() + "/" + username);
+		log.debug("Redirecting to {}", location);
+		return Optional.of(location);
+	}
+
+	public Optional<User> getUser(String username)
+	{
+		return
+				Optional
+						.ofNullable(streams.store(storeParameters).get(username))
+						.map(json ->
+						{
+							try
+							{
+								return mapper.readValue(json, User.class);
+							}
+							catch (JsonProcessingException e)
+							{
+								throw new RuntimeException(e);
+							}
+						});
+	}
+
+	@PostConstruct
+	public void start()
+	{
+		log.info("Starting Stream-Processor");
+		streams.start();
+	}
+
+	@PreDestroy
+	public void stop()
+	{
+		log.info("Stopping Stream-Processor");
+		streams.close();
+	}
+}