package com.shapira.examples.streams.wordcount;


import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.StreamsBuilder;
import java.util.Arrays;
import java.util.Properties;
import java.util.regex.Pattern;

public class WordCountExample {

    public static void main(String[] args) throws Exception{

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        // Ustawiamy reset przesunięcia na earliest, byśmy mogli ponownie uruchomić ten kod demo z tymi samymi wstępnie załadowanymi danymi
        // Uwaga: aby ponownie uruchomić demo, musisz użyć narzędzia do resetowania przesunięcia:
        // https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Obejście problemu momentu tworzenia wewnętrznych tematów
        // Błąd naprawiony w Kafce 0.10.2.0
        // Nie używaj w dużych aplikacjach produkcyjnych - zwiększa to obciążenie sieci
        // props.put(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, 500);

        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> source = builder.stream("wordcount-input");


        final Pattern pattern = Pattern.compile("\\W+");
        KStream counts  = source.flatMapValues(value-> Arrays.asList(pattern.split(value.toLowerCase())))
                .map((key, value) -> new KeyValue<Object, Object>(value, value))
                .filter((key, value) -> (!value.equals("the")))
                .groupByKey()
                .count().mapValues(value->Long.toString(value)).toStream();
        counts.to("wordcount-output");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // Dzięki temu działa resetowanie. Nie używaj w środowisku produkcyjnym - powoduje to, że przy każdym uruchomieniu aplikacja ponownie ładuje stan z Kafki
        streams.cleanUp();

        streams.start();

        // Zazwyczaj aplikacja przesyłania strumieniowego działa w nieskończoność,
        // ale w tym przykładzie pozwalamy jej działać przez jakiś czas i zatrzymujemy ją, ponieważ kończą się dane wejściowe
        Thread.sleep(5000L);

        streams.close();

    }
}
