package com.atlassian.stash.internal.hazelcast;

import com.atlassian.fugue.Option;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.StreamSerializer;

import java.io.IOException;

/**
 * Serializes Fugue's {@link Option}.
 * <p>
 * The primary motivator for the existence of this class is to allow the <i>data</i> inside an {@link Option} to
 * be serialized. For example, {@link com.atlassian.stash.internal.license.LicenseServiceImpl LicenseServiceImpl}
 * caches a {@link com.atlassian.stash.internal.license.DualLicense DualLicense} wrapped in an {@link Option}.
 * This serializer allows the {@code DualLicense} to be serialized by {@link LicenseStreamSerializer}.
 * <p>
 * Note: Future versions of Hazelcast are supposed to allow hijacking serialization even for fields inside normal
 * {@link java.io.Serializable Serializable} types, which would render this serializer unnecessary. However, they
 * have warned that <i>doing</i> so is very slow, so it may be desirable to retain this serializer anyway.
 *
 * @since 3.2
 */
public class OptionStreamSerializer implements StreamSerializer<Option<?>> {

    @Override
    public void destroy() {
    }

    @Override
    public int getTypeId() {
        return HazelcastConstants.TYPE_OPTION;
    }

    @Override
    public Option<?> read(ObjectDataInput in) throws IOException {
        boolean defined = in.readBoolean();

        return defined ? Option.some(in.readObject()) : Option.none();
    }

    @Override
    public void write(ObjectDataOutput out, Option<?> option) throws IOException {
        boolean defined = option.isDefined();
        out.writeBoolean(defined);
        if (defined) {
            out.writeObject(option.get());
        }
    }
}
