package com.example;
//
//import com.fasterxml.jackson.core.JsonProcessingException;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import org.apache.kafka.clients.producer.*;
//import org.apache.kafka.common.serialization.LongSerializer;
//import org.apache.kafka.common.serialization.StringSerializer;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.util.Properties;
//import java.util.concurrent.ExecutionException;
//
//public class MyProducer
//{
//    //private static final String TOPIC = "";
//    private static final String BOOTSTRAP_SERVERS = "192.168.181.136:9092";
//    private static final Logger log = LoggerFactory.getLogger(MyProducer.class);
//    public static boolean logTransaction(TransactionInfo transactionInfo , String topic)
//    {
//
//        log.info("Registering A Transaction for Mr/Ms : " + transactionInfo.getName());
//        ObjectMapper objectMapper = new ObjectMapper();
//
//        String value = null;
//        try
//        {
//            value = objectMapper.writeValueAsString(transactionInfo);
//        }
//        catch (JsonProcessingException e)
//        {
//            log.error("Transaction for Mr/Ms : " + transactionInfo.getName() + " Could Not Be Processed To Json");
//            return false;
//        }
//
//        Producer<Long, String> kafkaProducer = createKafkaProducer(BOOTSTRAP_SERVERS);
//
//        long timeStamp = System.currentTimeMillis();
//
//        ProducerRecord<Long , String> record = new ProducerRecord<>(topic , timeStamp ,value);
//
//        RecordMetadata recordMetadata = null;
//        try
//        {
//            recordMetadata = kafkaProducer.send(record).get();
//        }
//        catch (InterruptedException | ExecutionException e)
//        {
//            log.error("Could Not Send Data to Kafka , Or Could Not Get Meta-Data");
//            return false;
//        }
//
//
//        log.info((String.format("Record for Mr/Ms : %s -  with (key: %s, value: %s), " +
//                        "was sent to (partition: %d, offset: %d)",
//                transactionInfo.getName(),record.key(),
//                record.value(), recordMetadata.partition(),
//                recordMetadata.offset())));
//
//        System.out.println((String.format("Record for Mr/Ms : %s -  with (key: %s, value: %s), " +
//                        "was sent to (partition: %d, offset: %d)",
//                transactionInfo.getName(),record.key(),
//                record.value(), recordMetadata.partition(),
//                recordMetadata.offset())));
//        kafkaProducer.flush();
//        kafkaProducer.close();
//        return true;
//    }
//
//    public static Producer<Long, String> createKafkaProducer(String bootstrapServers) {
//        Properties properties = new Properties();
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "events-producer");
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
//        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//
//        return new KafkaProducer<>(properties);
//    }
//
//}
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class MyProducer {
    private static final String BOOTSTRAP_SERVERS = "192.168.181.136:9092";
    private static final Logger log = LoggerFactory.getLogger(MyProducer.class);
    public static final ObjectMapper objectMapper = new ObjectMapper();

    public static boolean logTransaction(TransactionInfo transactionInfo, String topic) {
        log.info("Registering a transaction for Mr/Ms: " + transactionInfo.getName());

        Serializer<TransactionInfo> valueSerializer = new TransactionInfoSerializer();
        byte[] valueBytes;
        try {
            valueBytes = valueSerializer.serialize(topic, transactionInfo);
        } catch (SerializationException e) {
            log.error("Transaction for Mr/Ms: " + transactionInfo.getName() + " could not be serialized");
            return false;
        }

        Producer<Long, TransactionInfo> kafkaProducer = createKafkaProducer(BOOTSTRAP_SERVERS);

        long timeStamp = System.currentTimeMillis();

        ProducerRecord<Long, TransactionInfo> record = new ProducerRecord<>(topic, timeStamp, transactionInfo);

        RecordMetadata recordMetadata;
        try {
            recordMetadata = kafkaProducer.send(record).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("Could not send data to Kafka or could not get metadata");
            return false;
        }

        log.info(String.format("Record for Mr/Ms: %s - with (key: %s, value: %s), " +
                        "was sent to (partition: %d, offset: %d)",
                transactionInfo.getName(), record.key(),
                record.value().toString(), recordMetadata.partition(),
                recordMetadata.offset()));

        System.out.println(String.format("Record for Mr/Ms: %s - with (key: %s, value: %s), " +
                        "was sent to (partition: %d, offset: %d)",
                transactionInfo.getName(), record.key(),
                record.value().toString(), recordMetadata.partition(),
                recordMetadata.offset()));

        kafkaProducer.flush();
        kafkaProducer.close();
        return true;
    }

    public static Producer<Long, TransactionInfo> createKafkaProducer(String bootstrapServers) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "events-producer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, TransactionInfoSerializer.class.getName());

        return new KafkaProducer<>(properties);
    }
}

