// Path: j1st-mqtt-communicator-kafka/src/main/java/com/github/j1stiot/mqtt/communicator/kafka/util/JSONs.java
// public class JSONs {
// 
//     // Global JSON ObjectMapper
//     public static final ObjectMapper Mapper = new ObjectMapper();
// 
//     static {
//         Mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
//         Mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//         Mapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
//         Mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
//     }
// 
//     private JSONs() {
//     }
// }


import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.JsonNode;
import com.github.j1stiot.mqtt.api.internal.*;
import com.github.j1stiot.mqtt.communicator.kafka.util.JSONs;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import java.io.IOException;
import java.util.Map;
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // nothing to do
    }

    @Override
    @SuppressWarnings("unchecked")
    public InternalMessage deserialize(String topic, byte[] bytes) {
        try {
            JavaType type = JSONs.Mapper.getTypeFactory().constructParametrizedType(InternalMessage.class, InternalMessage.class, JsonNode.class);
            InternalMessage m = JSONs.Mapper.readValue(bytes, type);
            switch (m.getMessageType()) {
                case CONNECT:
                    Connect connect = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Connect.class);
                    m.setPayload(connect);
                    return m;
                case CONNACK:
                    ConnAck connack = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), ConnAck.class);
                    m.setPayload(connack);
                    return m;
                case SUBSCRIBE:
                    Subscribe subscribe = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Subscribe.class);
                    m.setPayload(subscribe);
                    return m;
                case SUBACK:
                    SubAck suback = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), SubAck.class);
                    m.setPayload(suback);
                    return m;
                case UNSUBSCRIBE:
                    Unsubscribe unsubscribe = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Unsubscribe.class);
// Path: j1st-mqtt-communicator-kafka/src/main/java/com/github/j1stiot/mqtt/communicator/kafka/util/JSONs.java
// public class JSONs {
// 
//     // Global JSON ObjectMapper
//     public static final ObjectMapper Mapper = new ObjectMapper();
// 
//     static {
//         Mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
//         Mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//         Mapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
//         Mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
//     }
// 
//     private JSONs() {
//     }
// }


// Path: j1st-mqtt-communicator-kafka/src/main/java/com/github/j1stiot/mqtt/communicator/kafka/codec/InternalMessageDeserializer.java
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.JsonNode;
import com.github.j1stiot.mqtt.api.internal.*;
import com.github.j1stiot.mqtt.communicator.kafka.util.JSONs;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import java.io.IOException;
import java.util.Map;

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

    @Override
    @SuppressWarnings("unchecked")
    public InternalMessage deserialize(String topic, byte[] bytes) {
        try {
            JavaType type = JSONs.Mapper.getTypeFactory().constructParametrizedType(InternalMessage.class, InternalMessage.class, JsonNode.class);
            InternalMessage m = JSONs.Mapper.readValue(bytes, type);
            switch (m.getMessageType()) {
                case CONNECT:
                    Connect connect = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Connect.class);
                    m.setPayload(connect);
                    return m;
                case CONNACK:
                    ConnAck connack = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), ConnAck.class);
                    m.setPayload(connack);
                    return m;
                case SUBSCRIBE:
                    Subscribe subscribe = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Subscribe.class);
                    m.setPayload(subscribe);
                    return m;
                case SUBACK:
                    SubAck suback = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), SubAck.class);
                    m.setPayload(suback);
                    return m;
                case UNSUBSCRIBE:
                    Unsubscribe unsubscribe = JSONs.Mapper.treeToValue((JsonNode) m.getPayload(), Unsubscribe.class);
                    m.setPayload(unsubscribe);