import {Consumer, KafkaClient} from "kafka-node";
import config from "../config";
import Note from "../domain/Note";
import Sequence from "../domain/Sequence";
import Stimulus from "../domain/Stimulus";

export default class KafkaEventDelegate {
    public subscribers: any[];
    public client: KafkaClient;
    public consumer: Consumer;

    constructor() {
        this.subscribers = [];
        this.client = new KafkaClient({
            kafkaHost: config.kafka_server,
        });
        this.consumer = new Consumer(this.client,
            [],
            {
                autoCommit: true,
                fetchMaxWaitMs: 1000,
                encoding: "utf8",
                fromOffset: false,
            }
        );

        addTopicWithRetry(this.consumer);

        this.consumer.on("message", (message) => {

            this.subscribers.forEach((subscriber) => {
                try {
                    const data = JSON.parse(message.value.toString());

                    switch (data.kind) {
                        case "exploration": {
                            const exploration = data.exploration;
                            const seq = new Sequence();
                            exploration.forEach((interaction) => {
                                let value = interaction.kind;
                                if (interaction.value !== undefined) {
                                    value = value + "$" + interaction.value;
                                }
    
                                if (interaction.concreteType === "Action") {
                                    seq.addStimulus(new Stimulus(value));
                                } else if (interaction.concreteType === "Comment") {
                                    seq.addNote(new Note(value));
                                }
                            });
                            subscriber.hasNewSequence(data.sessionId, seq);
                            break;
                        }
                        default:
                            break;
                    }
                } catch (e) {
                    console.log('message is not JSON: ',e);
                }
            });
        });
    }

    public subscribe(subscriber): void {
        this.subscribers.push(subscriber);
    }
}

function addTopicWithRetry(consumer): void {
    consumer.addTopics(["MAIN_TOPIC"], (err) => {
        if (!err) {
            console.log("connected to kafka topic");
        } else {
            console.log("retrying to connect to kafka topic");
            const TIMEOUT = 3000;
            setTimeout(() => addTopicWithRetry(consumer), TIMEOUT);
        }
    });
}
