package io.quarkus.it.kafka.avro;

import java.time.Duration;

import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import io.vertx.core.json.JsonObject;

/**
 * Endpoint to test the Avro support
 */
@Path("/avro")
public class AvroEndpoint {

    @GET
    @Path("/confluent")
    public JsonObject getConfluent() {
        return get(AvroKafkaCreator.createConfluentConsumer("test-avro-confluent-consumer", "test-avro-confluent-consumer"));
    }

    @POST
    @Path("/confluent")
    public void sendConfluent(Pet pet) {
        KafkaProducer<Integer, Pet> p = AvroKafkaCreator.createConfluentProducer("test-avro-confluent");
        send(p, pet, "test-avro-confluent-producer");
    }

    @GET
    @Path("/apicurio")
    public JsonObject getApicurio() {
        return get(AvroKafkaCreator.createApicurioConsumer("test-avro-apicurio-consumer", "test-avro-apicurio-consumer"));
    }

    @POST
    @Path("/apicurio")
    public void sendApicurio(Pet pet) {
        KafkaProducer<Integer, Pet> p = AvroKafkaCreator.createApicurioProducer("test-avro-apicurio");
        send(p, pet, "test-avro-apicurio-producer");
    }

    private JsonObject get(KafkaConsumer<Integer, Pet> consumer) {
        final ConsumerRecords<Integer, Pet> records = consumer.poll(Duration.ofMillis(60000));
        if (records.isEmpty()) {
            return null;
        }
        ConsumerRecord<Integer, Pet> consumerRecord = records.iterator().next();
        Pet p = consumerRecord.value();
        // We cannot serialize the returned Pet directly, it contains non-serializable object such as the schema.
        JsonObject result = new JsonObject();
        result.put("name", p.getName());
        result.put("color", p.getColor());
        return result;
    }

    private void send(KafkaProducer<Integer, Pet> producer, Pet pet, String topic) {
        producer.send(new ProducerRecord<>(topic, 0, pet));
        producer.flush();
    }
}
