package ru.kayashov.bar.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Serializer;
import ru.kayashov.bar.model.AbstractSendMessage;

import java.util.Map;

@Slf4j
public class AbstractMessageSerializer implements Serializer<AbstractSendMessage> {

    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
    }

    @Override
    public byte[] serialize(String topic, AbstractSendMessage data) {
        try {
            if (data == null){
                log.warn("Отсутствует объект для отправки сообщения");
                return null;
            }
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new SerializationException("Ошибка сериализации объекта");
        }
    }

    @Override
    public byte[] serialize(String topic, Headers headers, AbstractSendMessage data) {
        return Serializer.super.serialize(topic, headers, data);
    }

    @Override
    public void close() {
    }
}