/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.query.schema.message;

import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.QueryIndexType;
import org.apache.ignite.internal.CoreMessagesProvider;
import org.apache.ignite.internal.direct.DirectMessageReader;
import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
import org.apache.ignite.internal.processors.query.QueryEntityEx;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;

import static org.apache.ignite.marshaller.Marshallers.jdk;

/** Serialization round-trip of {@link QueryEntityMessage} and {@link QueryEntityExMessage}. */
public class QueryEntityMessageSerializationTest extends GridCommonAbstractTest {
    /** Marshaller for the {@code dfltFieldValues} payload (matches the {@code withNoSchema} registration). */
    private final Marshaller marsh = jdk();

    /** Factory aware of {@link QueryEntityMessage} and its nested messages. */
    private final MessageFactory msgFactory = new IgniteMessageFactoryImpl(
        new MessageFactoryProvider[] {new CoreMessagesProvider(marsh, marsh, U.gridClassLoader())});

    /** */
    @Test
    public void testQueryEntity() throws Exception {
        QueryEntity entity = queryEntity();

        assertEquals(entity, roundTrip(entity));
    }

    /** */
    @Test
    public void testQueryEntityEx() throws Exception {
        QueryEntityEx entity = queryEntityEx();

        QueryEntity res = roundTrip(entity);

        assertTrue(res instanceof QueryEntityEx);
        assertEquals(entity, res);

        // Not part of QueryEntityEx.equals(), so assert it explicitly.
        assertEquals(entity.fillAbsentPKsWithDefaults(), ((QueryEntityEx)res).fillAbsentPKsWithDefaults());
    }

    /**
     * @param src Source entity.
     * @return Entity rebuilt after message conversion and a full direct-marshalling round-trip.
     */
    private QueryEntity roundTrip(QueryEntity src) throws IgniteCheckedException {
        QueryEntityMessage msg = src instanceof QueryEntityEx
            ? new QueryEntityExMessage((QueryEntityEx)src) : new QueryEntityMessage(src);

        msg.prepareMarshal(marsh);

        QueryEntityMessage res = doMarshalUnmarshal(msg);

        res.finishUnmarshal(marsh, U.gridClassLoader());

        return res.toEntity();
    }

    /**
     * @param msg Message to write and read back through {@link DirectMessageWriter}/{@link DirectMessageReader}.
     * @return Restored message.
     */
    private QueryEntityMessage doMarshalUnmarshal(QueryEntityMessage msg) {
        ByteBuffer buf = ByteBuffer.allocate(64 * 1024);

        DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
        writer.setBuffer(buf);

        assertTrue(writer.writeMessage(msg, false));

        buf.flip();

        DirectMessageReader reader = new DirectMessageReader(msgFactory, null);
        reader.setBuffer(buf);

        QueryEntityMessage res = (QueryEntityMessage)reader.readMessage(false);

        assertNotNull(res);

        return res;
    }

    /** @return Query entity with every field populated, including non-empty default field values. */
    private QueryEntity queryEntity() {
        LinkedHashMap<String, String> fields = new LinkedHashMap<>();
        fields.put("id", Integer.class.getName());
        fields.put("name", String.class.getName());
        fields.put("price", BigDecimal.class.getName());

        Map<String, Object> dfltVals = new HashMap<>();
        dfltVals.put("name", "unknown");
        dfltVals.put("price", new BigDecimal("9.99"));
        dfltVals.put("id", 42);

        Map<String, Integer> precision = new HashMap<>();
        precision.put("name", 64);

        Map<String, Integer> scale = new HashMap<>();
        scale.put("price", 2);

        Map<String, String> aliases = new HashMap<>();
        aliases.put("name", "NAME_ALIAS");

        return new QueryEntity()
            .setKeyType(Integer.class.getName())
            .setValueType("org.apache.ignite.Person")
            .setKeyFieldName("id")
            .setValueFieldName("name")
            .setTableName("PERSON")
            .setFields(fields)
            .setKeyFields(new LinkedHashSet<>(Arrays.asList("id")))
            .setAliases(aliases)
            .setIndexes(Collections.singletonList(new QueryIndex("name", QueryIndexType.SORTED).setInlineSize(32)))
            .setNotNullFields(new HashSet<>(Arrays.asList("id", "name")))
            .setDefaultFieldValues(dfltVals)
            .setFieldsPrecision(precision)
            .setFieldsScale(scale);
    }

    /** @return Extended query entity with all base and extended fields populated. */
    private QueryEntityEx queryEntityEx() {
        QueryEntity base = queryEntity();
        base.setNotNullFields(null);

        QueryEntityEx entity = new QueryEntityEx(base);

        // Natural QueryEntityEx: notNullFields lives only in the Ex field, base shadow stays null.
        entity.setNotNullFields(new HashSet<>(Arrays.asList("id", "name")));
        entity.setPreserveKeysOrder(true);
        entity.implicitPk(true);
        entity.fillAbsentPKsWithDefaults(true);
        entity.sql(true);
        entity.setPrimaryKeyInlineSize(16);
        entity.setAffinityKeyInlineSize(24);

        return entity;
    }
}