package com.shapira.examples.streams.stockstats;

import com.shapira.examples.streams.stockstats.serde.JsonSerializer;
import com.shapira.examples.streams.stockstats.model.Trade;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;


public class StockGenProducer {

    public static KafkaProducer<String, Trade> producer = null;

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

        System.out.println("Aby zatrzymać generowanie danych, naciśnij CTRL+C");


        // Dodajemy zaczep zamykania
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                System.out.println("Zamykanie");
                if (producer != null)
                    producer.close();
            }
        });

        JsonSerializer<Trade> tradeSerializer = new JsonSerializer<>();

        // Konfigurujemy producent
        Properties props;
        if (args.length==1)
            props = LoadConfigs.loadConfig(args[0]);
        else
            props = LoadConfigs.loadConfig();

        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", tradeSerializer.getClass().getName());

        // Uruchamiamy producent
        producer = new KafkaProducer<>(props);


        // Inicjujemy

        Random random = new Random();
        long iter = 0;

        Map<String, Integer> prices = new HashMap<>();

        for (String ticker : Constants.TICKERS)
            prices.put(ticker, Constants.START_PRICE);

        // Uruchamiamy generowanie zdarzeń; aby zatrzymać, naciścij CTRL+C

        while (true) {
            iter++;
            for (String ticker : Constants.TICKERS) {
                double log = random.nextGaussian() * 0.25 + 1; // Losowa zmienna z rozkładu logarytmicznie normalnego z odchyleniem standardowym=0,25 i średnią=1
                int size = random.nextInt(100);
                int price = prices.get(ticker);

                // Czasami ceny się wahają
                if (iter % 10 == 0) {
                    price = price + random.nextInt(Constants.MAX_PRICE_CHANGE * 2) - Constants.MAX_PRICE_CHANGE;
                    prices.put(ticker, price);
                }

                Trade trade = new Trade("ASK",ticker,(price+log),size);
                // Zwróć uwagę, że jako klucza używamy symbolu spółki, aby wszystkie ceny ASK dla tych samych akcji znajdowały się w jednej partycji
                ProducerRecord<String, Trade> record = new ProducerRecord<>(Constants.STOCK_TOPIC, ticker, trade);

                producer.send(record, (RecordMetadata r, Exception e) -> {
                    if (e != null) {
                        System.out.println("Błąd podczas produkcji zdarzeń");
                        e.printStackTrace();
                    }
                });

                // Uśpienie na chwilę; inaczej usmaży Ci maszynę
                Thread.sleep(Constants.DELAY);
            }
        }
    }
}
