package com.esotericsoftware.kryo.serializers;

import static com.esotericsoftware.kryo.util.Util.className;
import static com.esotericsoftware.kryo.util.Util.pos;
import static com.esotericsoftware.minlog.Log.DEBUG;
import static com.esotericsoftware.minlog.Log.TRACE;
import static com.esotericsoftware.minlog.Log.debug;
import static com.esotericsoftware.minlog.Log.trace;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.Registration;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.InputChunked;
import com.esotericsoftware.kryo.util.Util;

/**
 * 解决KRYO <= 5.2.0原生实现针对存在默认值字段，如果设值为null时，反序列化后依旧为默认值问题</br>
 * 
 * @see https://github.com/EsotericSoftware/kryo/issues/851
 * 
 * @author Kolor
 * @date 2021-8-18 12:10:11
 * @param <T>
 */
@SuppressWarnings("rawtypes")
public class DefaultValueSetNullSupportCompatibleFieldSerializer<T> extends CompatibleFieldSerializer<T> {
    private static final int                      binarySearchThreshold = 32;
    private final CompatibleFieldSerializerConfig config;

    public DefaultValueSetNullSupportCompatibleFieldSerializer(Kryo kryo, Class type) {
        this(kryo, type, new CompatibleFieldSerializerConfig());
    }

    public DefaultValueSetNullSupportCompatibleFieldSerializer(Kryo kryo, Class type, CompatibleFieldSerializerConfig config) {
        super(kryo, type, config);
        this.config = config;
    }

    private void setNullValue(CachedField cachedField, Object obj) {
        if (!cachedField.getField().getType().isPrimitive()) {
            try {
                cachedField.getField().set(obj, null);
            } catch (IllegalArgumentException | IllegalAccessException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public T read(Kryo kryo, Input input, Class<? extends T> type) {
        int pop = pushTypeVariables();

        T object = create(kryo, input, type);
        kryo.reference(object);

        CachedField[] fields = (CachedField[]) kryo.getGraphContext().get(this);
        if (fields == null) fields = readFields(kryo, input);

        boolean chunked = config.chunked, readUnknownTagData = config.readUnknownFieldData;
        Input fieldInput;
        InputChunked inputChunked = null;
        if (chunked)
            fieldInput = inputChunked = new InputChunked(input, config.chunkSize);
        else
            fieldInput = input;
        for (int i = 0, n = fields.length; i < n; i++) {
            CachedField cachedField = fields[i];

            if (readUnknownTagData) {
                Registration registration;
                try {
                    registration = kryo.readClass(fieldInput);
                } catch (KryoException ex) {
                    String message = "Unable to read unknown data (unknown type). (" + getType().getName() + "#" + cachedField + ")";
                    if (!chunked) throw new KryoException(message, ex);
                    if (DEBUG) debug("kryo", message, ex);
                    inputChunked.nextChunk();
                    continue;
                }
                if (registration == null) {
                    // 修复存在默认值时，而当前值为null时，未设置的问题
                    setNullValue(cachedField, object);

                    if (chunked) inputChunked.nextChunk();
                    continue;
                }
                Class valueClass = registration.getType();
                if (cachedField == null) {
                    // Read unknown data in case it is a reference.
                    if (TRACE) trace("kryo", "Read unknown data, type: " + className(valueClass) + pos(input.position()));
                    try {
                        kryo.readObject(fieldInput, valueClass);
                    } catch (KryoException ex) {
                        String message = "Unable to read unknown data, type: " + className(valueClass) + " (" + getType().getName()
                                + "#" + cachedField + ")";
                        if (!chunked) throw new KryoException(message, ex);
                        if (DEBUG) debug("kryo", message, ex);
                    }
                    if (chunked) inputChunked.nextChunk();
                    continue;
                }

                // Ensure the type in the data is compatible with the field type.
                if (cachedField.valueClass != null && !Util.isAssignableTo(valueClass, cachedField.field.getType())) {
                    String message = "Read type is incompatible with the field type: " + className(valueClass) + " -> "
                            + className(cachedField.valueClass) + " (" + getType().getName() + "#" + cachedField + ")";
                    if (!chunked) throw new KryoException(message);
                    if (DEBUG) debug("kryo", message);
                    inputChunked.nextChunk();
                    continue;
                }

                cachedField.setCanBeNull(false);
                cachedField.setValueClass(valueClass);
                cachedField.setReuseSerializer(false);
            } else if (cachedField == null) {
                if (!chunked) throw new KryoException("Unknown field. (" + getType().getName() + ")");
                if (TRACE) trace("kryo", "Skip unknown field.");
                inputChunked.nextChunk();
                continue;
            }

            if (TRACE) log("Read", cachedField, input.position());
            cachedField.read(fieldInput, object);
            if (chunked) inputChunked.nextChunk();
        }

        popTypeVariables(pop);
        return object;
    }

    private CachedField[] readFields(Kryo kryo, Input input) {
        if (TRACE) trace("kryo", "Read fields for class: " + type.getName());

        int length = input.readVarInt(true);
        String[] names = new String[length];
        for (int i = 0; i < length; i++) {
            names[i] = input.readString();
            if (TRACE) trace("kryo", "Read field name: " + names[i]);
        }

        CachedField[] fields = new CachedField[length];
        CachedField[] allFields = cachedFields.fields;
        if (length < binarySearchThreshold) {
            outer: for (int i = 0; i < length; i++) {
                String schemaName = names[i];
                for (int ii = 0, nn = allFields.length; ii < nn; ii++) {
                    if (allFields[ii].name.equals(schemaName)) {
                        fields[i] = allFields[ii];
                        continue outer;
                    }
                }
                if (TRACE) trace("kryo", "Unknown field will be skipped: " + schemaName);
            }
        } else {
            int low, mid, high, compare;
            int lastFieldIndex = allFields.length - 1;
            outer: for (int i = 0; i < length; i++) {
                String schemaName = names[i];
                low = 0;
                high = lastFieldIndex;
                while (low <= high) {
                    mid = (low + high) >>> 1;
                    compare = schemaName.compareTo(allFields[mid].name);
                    if (compare < 0)
                        high = mid - 1;
                    else if (compare > 0)
                        low = mid + 1;
                    else {
                        fields[i] = allFields[mid];
                        continue outer;
                    }
                }
                if (TRACE) trace("kryo", "Unknown field will be skipped: " + schemaName);
            }
        }

        kryo.getGraphContext().put(this, fields);
        return fields;
    }

}

if (registration == null) {
	// 修复存在默认值时，而当前值为null时，未设置的问题
	if (cachedField != null) {
		setNullValue(cachedField, object);
	}

	if (chunked) inputChunked.nextChunk();
	continue;
}