package com.kevinsimard.kafka.streams;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.connect.json.JsonDeserializer;
import org.apache.kafka.connect.json.JsonSerializer;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Properties;

public class Aggregator {

    private static final String KAFKA_HOSTS =
        Optional.of(System.getenv("KAFKA_HOSTS")).orElse("localhost:9092");

    private static final Serde<JsonNode> JSON_SERDE =
        Serdes.serdeFrom(new JsonSerializer(), new JsonDeserializer());

    private static List<String> processedMessages = new ArrayList<>();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 0);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "aggregator");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_HOSTS);

        KafkaStreams streams = new KafkaStreams(buildStreams(), props);

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

        streams.start();
    }

    private static KStreamBuilder buildStreams() {
        KStreamBuilder builder = new KStreamBuilder();

        KStream<String, JsonNode> salesAggregated =
            builder.stream(Serdes.String(), JSON_SERDE, "sales-aggregated");

        salesAggregated.foreach(Aggregator::markAsProcessedInDatastore);

        KStream<String, JsonNode> salesRaw =
            builder.stream(Serdes.String(), JSON_SERDE, "sales-raw");

        salesRaw.filterNot(Aggregator::isAlreadyProcessed)
            .groupBy(Aggregator::groupByUserId, Serdes.String(), JSON_SERDE)
            .aggregate(() -> null, Aggregator::aggregateValues, JSON_SERDE, "aggregated-interm")
            .to(Serdes.String(), JSON_SERDE, "sales-aggregated");

        return builder;
    }

    @SuppressWarnings("unused")
    private static void markAsProcessedInCache(String key, JsonNode value) {
        String hash = uniqueHashForMessage(value);

        processedMessages.add(hash);
    }

    @SuppressWarnings("unused")
    private static void markAsProcessedInDatastore(String key, JsonNode value) {
        // TODO: Save the generated hash in datastore.

        // TODO: Uncomment when using datastore to prevent array from getting too big.
        //processedMessages.remove(uniqueHashForMessage(value));
    }

    private static Boolean isAlreadyProcessed(String key, JsonNode value) {
        // TODO: Also check in datastore if not found in cache.

        String hash = uniqueHashForMessage(value);
        Boolean isProcessed = processedMessages.indexOf(hash) != -1;

        if (! isProcessed) markAsProcessedInCache(key, value);

        return isProcessed;
    }

    @SuppressWarnings("unused")
    private static String groupByUserId(String key, JsonNode value) {
        return value.get("user_id").asText();
    }

    @SuppressWarnings("unused")
    private static JsonNode aggregateValues(String key, JsonNode value, JsonNode previous) {
        if (previous != null) {
            ((ObjectNode) value).put("total",
                previous.get("total").asDouble() +
                value.get("total").asDouble());
        }

        return value;
    }

    private static String uniqueHashForMessage(JsonNode value) {
        // TODO: Replace the following with a hash function.
        return value.get("user_id").asText() +
            ":" + value.get("sale_id").asText();
    }
}

@@ -1009,16 +1009,22 @@
  static 
+final 
 List%3CStr
@@ -1064,24 +1064,65 @@
 yList%3C%3E();%0A%0A
+    private final KafkaStreams streams;%0A%0A
     public s
@@ -1485,28 +1485,103 @@
         
-KafkaStreams
+new Aggregator(props).start();%0A    %7D%0A%0A    public Aggregator(Properties props) %7B%0A       
  streams
@@ -1624,17 +1624,16 @@
 props);%0A
-%0A
         
@@ -1694,24 +1694,56 @@
 s::close));%0A
+    %7D%0A%0A    public void start() %7B
 %0A        str
@@ -1775,23 +1775,16 @@
 private 
-static 
 KStreamB
@@ -3816,32 +3816,16 @@
 uble() +
-%0A               
  value.g
@@ -4061,28 +4061,16 @@
 Text() +
-%0A           
  %22:%22 + v

Add Aggregator class constructor