/*
 * Copyright (c) 2008-2015, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed 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 com.hazelcast.internal.serialization.impl;


import com.hazelcast.logging.Logger;
import com.hazelcast.nio.ClassLoaderUtil;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.DataSerializable;
import com.hazelcast.nio.serialization.DataSerializableFactory;
import com.hazelcast.internal.serialization.DataSerializerHook;
import com.hazelcast.nio.serialization.HazelcastSerializationException;
import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
import com.hazelcast.nio.serialization.StreamSerializer;
import com.hazelcast.util.ExceptionUtil;
import com.hazelcast.util.ServiceLoader;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import static com.hazelcast.internal.serialization.impl.SerializationConstants.CONSTANT_TYPE_DATA_SERIALIZABLE;

/**
 * This class is the default serializer for all types that are serialized using Hazelcast
 * internal methods. Due to the operation responding on deserialization errors this class
 * has a dependency to {@link com.hazelcast.nio.IOUtil#extractOperationCallId(com.hazelcast.nio.serialization.Data,
 * com.hazelcast.internal.serialization.SerializationService)}.
 * If the way the DataSerializer serializes values is changed the extract method needs to be changed too!
 */
final class DataSerializer implements StreamSerializer<DataSerializable> {

    private static final String FACTORY_ID = "com.hazelcast.DataSerializerHook";

    private final Map<Integer, DataSerializableFactory> factories = new HashMap<Integer, DataSerializableFactory>();

    DataSerializer(Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories, ClassLoader classLoader) {
        try {
            final Iterator<DataSerializerHook> hooks = ServiceLoader.iterator(DataSerializerHook.class, FACTORY_ID, classLoader);
            while (hooks.hasNext()) {
                DataSerializerHook hook = hooks.next();
                final DataSerializableFactory factory = hook.createFactory();
                if (factory != null) {
                    register(hook.getFactoryId(), factory);
                }
            }
        } catch (Exception e) {
            throw ExceptionUtil.rethrow(e);
        }

        if (dataSerializableFactories != null) {
            for (Map.Entry<Integer, ? extends DataSerializableFactory> entry : dataSerializableFactories.entrySet()) {
                register(entry.getKey(), entry.getValue());
            }
        }
    }

    private void register(int factoryId, DataSerializableFactory factory) {
        final DataSerializableFactory current = factories.get(factoryId);
        if (current != null) {
            if (current.equals(factory)) {
                Logger.getLogger(getClass()).warning("DataSerializableFactory[" + factoryId + "] is already registered! Skipping "
                        + factory);
            } else {
                throw new IllegalArgumentException("DataSerializableFactory[" + factoryId + "] is already registered! "
                        + current + " -> " + factory);
            }
        } else {
            factories.put(factoryId, factory);
        }
    }

    @Override
    public int getTypeId() {
        return CONSTANT_TYPE_DATA_SERIALIZABLE;
    }

    @Override
    public DataSerializable read(ObjectDataInput in) throws IOException {
        final DataSerializable ds;
        final boolean identified = in.readBoolean();
        int id = 0;
        int factoryId = 0;
        String className = null;
        try {
            // If you ever change the way this is serialized think about to change
            // BasicOperationService::extractOperationCallId
            if (identified) {
                factoryId = in.readInt();
                final DataSerializableFactory dsf = factories.get(factoryId);
                if (dsf == null) {
                    throw new HazelcastSerializationException("No DataSerializerFactory registered for namespace: " + factoryId);
                }
                id = in.readInt();
                ds = dsf.create(id);
                if (ds == null) {
                    throw new HazelcastSerializationException(dsf
                            + " is not be able to create an instance for id: " + id + " on factoryId: " + factoryId);
                }
                // TODO: @mm - we can check if DS class is final.
            } else {
                className = in.readUTF();
                ds = ClassLoaderUtil.newInstance(in.getClassLoader(), className);
            }
            ds.readData(in);
            return ds;
        } catch (Exception e) {
            if (e instanceof IOException) {
                throw (IOException) e;
            }
            if (e instanceof HazelcastSerializationException) {
                throw (HazelcastSerializationException) e;
            }
            throw new HazelcastSerializationException("Problem while reading DataSerializable, namespace: "
                    + factoryId
                    + ", id: " + id
                    + ", class: '" + className + "'"
                    + ", exception: " + e.getMessage(), e);
        }
    }

    @Override
    public void write(ObjectDataOutput out, DataSerializable obj) throws IOException {
        // If you ever change the way this is serialized think about to change
        // BasicOperationService::extractOperationCallId
        final boolean identified = obj instanceof IdentifiedDataSerializable;
        out.writeBoolean(identified);
        if (identified) {
            final IdentifiedDataSerializable ds = (IdentifiedDataSerializable) obj;
            out.writeInt(ds.getFactoryId());
            out.writeInt(ds.getId());
        } else {
            out.writeUTF(obj.getClass().getName());
        }
        obj.writeData(out);
    }

    @Override
    public void destroy() {
        factories.clear();
    }
}

@@ -1304,61 +1304,91 @@
 er;%0A
-%0A
 import 
-java.io.IOException;%0Aimport java.util.HashMap
+com.hazelcast.util.collection.Int2ObjectHashMap;%0A%0Aimport java.io.IOException
 ;%0Aim
@@ -2198,29 +2198,34 @@
 e final 
-Map%3CInteger, 
+Int2ObjectHashMap%3C
 DataSeri
@@ -2261,25 +2261,26 @@
 new 
-HashMap%3CInteger, 
+Int2ObjectHashMap%3C
 Data

Store IdentifiedDataSerializable factories in a specialized HashMap.