package fr.faraphel.m1_pe_kafka

/**
 * The consumer
 */

import org.apache.kafka.clients.consumer.*

import java.time.Duration
import java.util.Collections
import java.util.Properties


/**
 * The consumer
 */
class Consumer {
    private val properties: Properties = Properties()
    private var consumer: KafkaConsumer<String, String>
    private val topic: String = "mon_beau_topic"

    init {
        this.properties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringSerializer"
        this.properties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringSerializer"
        this.properties[ConsumerConfig.GROUP_ID_CONFIG] = "premier_groupe"
        this.properties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
        this.properties[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = System.getenv("KAFKA_BOOTSTRAP_SERVERS")
            ?: throw IllegalArgumentException("Missing environment variable: KAFKA_BOOTSTRAP_SERVERS")

        this.consumer = KafkaConsumer(properties);
    }

    fun main() {
        this.consumer.subscribe(Collections.singletonList(topic));

        // Afficher les messages reçu
        while (true) {
            val messages: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(500));
            
            messages.forEach { message ->
                println(">" + message.value())
            }
        }
    }
}