public final class SnowcastEpoch {

    private static final long INITIALIZATION_TIMESTAMP = System.currentTimeMillis();
    private static final long INITIALIZATION_NANOTIME = System.nanoTime();

    private final long offset;

    private SnowcastEpoch(@Nonnegative long offset) {
        this.offset = offset;
    }

    /**
     * Returns the current point in time as a timestamp based on this custom epoch.
     *
     * @return a current timestamp based on this custom epoch
     */
    public long getEpochTimestamp() {
        return getEpochTimestamp(getNow());
    }

    /**
     * Returns the timestamp value between this custom epoch and the standard Linux
     * timestamp epoch (in milliseconds) for the given timestamp.
     *
     * @param timestamp a timestamp based on the standard Linux timestamp epoch
     * @return a timestamp based on this custom epoch
     */
    long getEpochTimestamp(@Nonnegative long timestamp) {
        return timestamp - offset;
    }

    /**
     * Returns the configured offset of this custom epoch. This value is the number
     * of milliseconds between the standard Linux timestamp (in milliseconds) and the
     * configured date / time of this epoch.
     *
     * @return the number of milliseconds (offset) from the standard Linux timestamp
     */
    public long getEpochOffset() {
        return offset;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }

        SnowcastEpoch epoch = (SnowcastEpoch) o;
        return offset == epoch.offset;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return (int) (offset ^ (offset >>> 32));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return "SnowcastEpoch{" + "offset=" + offset + '}';
    }

    /**
     * Creates a custom epoch based on the given {@link java.util.Calendar} instance.
     *
     * @param calendar the calendar instance to base the epoch on
     * @return a SnowcastEpoch instance based on the given calendars value
     * @deprecated Replaced by {@link #byInstant(Instant)} 
     */
    public static SnowcastEpoch byCalendar(@Nonnull Calendar calendar) {
        calendar.set(Calendar.MILLISECOND, 0);
        long offset = calendar.getTimeInMillis();
        SnowcastEpoch epoch = new SnowcastEpoch(offset);

        if (epoch.getEpochTimestamp() < 0) {
            throw exception(ILLEGAL_CALENDAR_INSTANCE, calendar.getTime());
        }
        return epoch;
    }

    /**
     * Creates a custom epoch based on the given milliseconds based timestamp. This given
     * timestamp must be based on the standard Linux timestamp epoch (in milliseconds).
     *
     * @param timestamp the timestamp value to base the epoch on
     * @return a SnowcastEpoch instance based on the given timestamp value
     */
    @Nonnull
    public static SnowcastEpoch byTimestamp(@Nonnegative long timestamp) {
        return new SnowcastEpoch(timestamp);
    }

    /**
     * Creates a custom epoch based on the given instant. 
     *
     * @param instant the instant value to base the epoch on
     * @return a SnowcastEpoch instance based on the given timestamp value
     */
    @Nonnull
    public static SnowcastEpoch byInstant(@Nonnull Instant instant) {
        long offset = instant.toEpochMilli();
        SnowcastEpoch epoch = new SnowcastEpoch(offset);
        
        if (epoch.getEpochTimestamp() < 0) {
            throw exception(ILLEGAL_INSTANT, instant);
        }
        return epoch;
    }

    private static long getNow() {
        long nanoTime = System.nanoTime();
        long delta = nanoTime - INITIALIZATION_NANOTIME;
        return INITIALIZATION_TIMESTAMP + TimeUnit.NANOSECONDS.toMillis(delta);
    }
} public class SnowcastIllegalStateException
        extends SnowcastException {

    /**
     * Basic exception constructor
     *
     * @param message message to be used
     */
    public SnowcastIllegalStateException(@Nonnull String message) {
        super(message);
    }
} public enum SnowcastSequenceState {

    /**
     * A {@link com.noctarius.snowcast.SnowcastSequencer} in the state <tt>Detached</tt>
     * is not destroyed but <b>cannot</b> be used to generate IDs since there is no
     * logical node id assigned.
     */
    Detached,

    /**
     * A {@link com.noctarius.snowcast.SnowcastSequencer} in the state <tt>Attached</tt>
     * has a logical node id assigned and can be used to generate IDs.<br>
     * This is the <b>default</b> state after acquiring the sequencer for the first time!
     */
    Attached,

    /**
     * A {@link com.noctarius.snowcast.SnowcastSequencer} in the state <tt>Destroyed</tt>
     * does not have a legal configuration anymore. This instance <b>can never ever</b> be
     * used again to generate IDs. A sequencer with the same referral name might be created
     * again at that point.
     */
    Destroyed
} @ThreadSafe
public interface SnowcastSequencer {

    /**
     * Returns the reference name of the sequencer. This name is shared in the Hazelcast
     * cluster. Whenever the same sequencer name is referenced when creating a
     * SnowcastSequencer the same internal sequencer instance is returned (unless destroyed).
     *
     * @return the sequencers reference name
     */
    @Nonnull
    String getSequencerName();

    /**
     * <p>Generates and returns the next snowcast sequence ID. This call blocks until a ID is
     * free to be generated. The call blocks whenever a millisecond on the current node runs
     * out of possible IDs by exceeding the maximum counter value.</p>
     * <p>This method is fully thread-safe and expected to be used in multi-threading
     * environments.</p>
     *
     * @return the next available snowcast sequence ID
     * @throws InterruptedException if the blocking wait is interrupted
     */
    long next()
            throws InterruptedException;

    /**
     * Returns the current {@link com.noctarius.snowcast.SnowcastSequenceState} of this
     * sequencer. It might be attached, detached or destroyed. Please refer to the state
     * documentations to find out more about their meanings.
     *
     * @return the sequencers current state
     */
    @Nonnull
    SnowcastSequenceState getSequencerState();

    /**
     * Attaches the current SnowcastSequencer instance to a logical node ID. This is
     * a cluster operation and blocks until the owner of the reference name assigns
     * a logical node ID and returns it back to the current node or client.
     *
     * @return the current SnowcastSequencer instance for chaining of method calls
     */
    @Nonnull
    SnowcastSequencer attachLogicalNode();

    /**
     * Detaches the current SnowcastSequencer instance from the currently assigned
     * logical node ID. This is a cluster operation and blocks until the owner of
     * the reference name removed the logical node ID assignment and returns back
     * to the current node or client.
     *
     * @return the current SnowcastSequencer instance for chaining of method calls
     */
    @Nonnull
    SnowcastSequencer detachLogicalNode();

    /**
     * Extracts the timestamp value of the given snowcast sequence ID based on the
     * current configuration. Extracted timestamps are based on the configured
     * {@link com.noctarius.snowcast.SnowcastEpoch}. To get the real world point in
     * time the epoch value plus this returned timestamp must be summed together.
     *
     * @param sequenceId the sequence ID to extract the timestamp from
     * @return the extracted timestamp based on the configured epoch
     */
    @Nonnegative
    long timestampValue(long sequenceId);

    /**
     * Extracts the logical node ID of the given snowcast sequence ID based on the
     * current configuration. The extracted logical node ID is based no the configured
     * maximal logical node ID (number of bits used).
     *
     * @param sequenceId the sequence ID to extract the logical node ID from
     * @return the extracted logical node ID based on the configured maximal logical node ID
     */
    @Nonnegative
    int logicalNodeId(long sequenceId);

    /**
     * Extracts the counter value of the given snowcast sequence ID based on the
     * current configuration. The extracted counter value is based no the configured
     * maximal logical node ID (number of bits used).
     *
     * @param sequenceId the sequence ID to extract the counter value from
     * @return the extracted counter value based on the configured maximal logical node ID
     */
    @Nonnegative
    int counterValue(long sequenceId);
} public class SnowcastSequencerAlreadyRegisteredException
        extends SnowcastException {

    /**
     * Basic exception constructor
     *
     * @param message message to be used
     */
    public SnowcastSequencerAlreadyRegisteredException(@Nonnull String message) {
        super(message);
    }
} public class AttachLogicalNodeOperation
        extends AbstractSequencerOperation
        implements PartitionAwareOperation, BackupAwareOperation {

    private transient Integer logicalNodeId;

    private SequencerDefinition definition;

    public AttachLogicalNodeOperation() {
    }

    public AttachLogicalNodeOperation(SequencerDefinition definition) {
        super(definition.getSequencerName());
        this.definition = definition;
    }

    @Override
    public int getId() {
        return SequencerDataSerializerHook.TYPE_ATTACH_LOGICAL_NODE;
    }

    @Override
    public void run()
            throws Exception {

        NodeSequencerService sequencerService = getService();
        SequencerPartition partition = sequencerService.getSequencerPartition(getPartitionId());
        logicalNodeId = partition.attachLogicalNode(definition, getCallerAddress());
    }

    @Override
    public boolean returnsResponse() {
        return true;
    }

    @Override
    public Object getResponse() {
        return logicalNodeId;
    }

    @Override
    protected void writeInternal(ObjectDataOutput out)
            throws IOException {

        super.writeInternal(out);
        out.writeLong(definition.getEpoch().getEpochOffset());
        out.writeInt(definition.getMaxLogicalNodeCount());
        out.writeShort(definition.getBackupCount());
    }

    @Override
    protected void readInternal(ObjectDataInput in)
            throws IOException {

        super.readInternal(in);

        long epochOffset = in.readLong();
        int maxLogicalNodeCount = in.readInt();
        short backupCount = in.readShort();

        SnowcastEpoch epoch = SnowcastEpoch.byTimestamp(epochOffset);
        definition = new SequencerDefinition(getSequencerName(), epoch, maxLogicalNodeCount, backupCount);
    }

    @Override
    public boolean shouldBackup() {
        return true;
    }

    @Override
    public int getSyncBackupCount() {
        return definition.getBackupCount();
    }

    @Override
    public int getAsyncBackupCount() {
        return 0;
    }

    @Override
    public Operation getBackupOperation() {
        return new BackupAttachLogicalNodeOperation(definition, logicalNodeId, getCallerAddress());
    }
} public class CreateSequencerDefinitionOperation
        extends AbstractSequencerOperation
        implements BackupAwareOperation {

    private transient SequencerDefinition response;

    private SequencerDefinition definition;

    public CreateSequencerDefinitionOperation() {
    }

    public CreateSequencerDefinitionOperation(SequencerDefinition definition) {
        super(definition.getSequencerName());
        this.definition = definition;
    }

    @Override
    public int getId() {
        return SequencerDataSerializerHook.TYPE_CREATE_SEQUENCER_DEFINITION;
    }

    @Override
    public void run()
            throws Exception {

        NodeSequencerService sequencerService = getService();
        response = sequencerService.registerSequencerDefinition(definition);
    }

    @Override
    public boolean returnsResponse() {
        return true;
    }

    @Override
    public Object getResponse() {
        return response;
    }

    @Override
    protected void writeInternal(ObjectDataOutput out)
            throws IOException {

        super.writeInternal(out);
        out.writeLong(definition.getEpoch().getEpochOffset());
        out.writeInt(definition.getMaxLogicalNodeCount());
        out.writeShort(definition.getBackupCount());
    }

    @Override
    protected void readInternal(ObjectDataInput in)
            throws IOException {

        super.readInternal(in);

        long epochOffset = in.readLong();
        int maxLogicalNodeCount = in.readInt();
        short backupCount = in.readShort();

        SnowcastEpoch epoch = SnowcastEpoch.byTimestamp(epochOffset);
        definition = new SequencerDefinition(getSequencerName(), epoch, maxLogicalNodeCount, backupCount);
    }

    @Override
    public boolean shouldBackup() {
        return response != null;
    }

    @Override
    public int getSyncBackupCount() {
        return response.getBackupCount();
    }

    @Override
    public int getAsyncBackupCount() {
        return 0;
    }

    @Override
    public Operation getBackupOperation() {
        return new BackupCreateSequencerDefinitionOperation(response);
    }
} public class DestroySequencerDefinitionOperation
        extends AbstractSequencerOperation
        implements BackupAwareOperation {

    private transient int backupCount;

    public DestroySequencerDefinitionOperation() {
    }

    public DestroySequencerDefinitionOperation(String sequencerName) {
        super(sequencerName);
    }

    @Override
    public int getId() {
        return SequencerDataSerializerHook.TYPE_DESTROY_SEQUENCER_DEFINITION;
    }

    @Override
    public void run()
            throws Exception {

        NodeSequencerService sequencerService = getService();
        String sequencerName = getSequencerName();

        SequencerDefinition definition = sequencerService.destroySequencer(sequencerName, true);

        // Definition might be already destroyed concurrently
        if (definition == null) {
            return;
        }

        backupCount = definition.getBackupCount();

        NodeEngine nodeEngine = getNodeEngine();
        OperationService operationService = nodeEngine.getOperationService();

        DestroySequencerOperation operation = new DestroySequencerOperation(sequencerName);
        for (MemberImpl member : nodeEngine.getClusterService().getMemberImpls()) {
            if (!member.localMember() && !member.getAddress().equals(getCallerAddress())) {
                operationService.invokeOnTarget(SERVICE_NAME, operation, member.getAddress());
            }
        }

        ClientDestroySequencerNotification notification = new ClientDestroySequencerNotification(sequencerName);
        Collection<EventRegistration> registrations = sequencerService.findClientChannelRegistrations(sequencerName, null);
        nodeEngine.getEventService().publishEvent(SERVICE_NAME, registrations, notification, 1);
    }

    @Override
    public boolean returnsResponse() {
        return true;
    }

    @Override
    public Object getResponse() {
        return Boolean.TRUE;
    }

    @Override
    public boolean shouldBackup() {
        return true;
    }

    @Override
    public int getSyncBackupCount() {
        return backupCount;
    }

    @Override
    public int getAsyncBackupCount() {
        return 0;
    }

    @Override
    public Operation getBackupOperation() {
        return new BackupDestroySequencerDefinitionOperation(getSequencerName());
    }
} public class DetachLogicalNodeOperation
        extends AbstractSequencerOperation
        implements PartitionAwareOperation, BackupAwareOperation {

    private int logicalNodeId;
    private SequencerDefinition definition;

    public DetachLogicalNodeOperation() {
    }

    public DetachLogicalNodeOperation(SequencerDefinition definition, int logicalNodeId) {
        super(definition.getSequencerName());
        this.definition = definition;
        this.logicalNodeId = logicalNodeId;
    }

    @Override
    public int getId() {
        return SequencerDataSerializerHook.TYPE_DETACH_LOGICAL_NODE;
    }

    @Override
    public void run()
            throws Exception {

        NodeSequencerService sequencerService = getService();
        SequencerPartition partition = sequencerService.getSequencerPartition(getPartitionId());
        partition.detachLogicalNode(getSequencerName(), getCallerAddress(), logicalNodeId);
    }

    @Override
    public boolean returnsResponse() {
        return true;
    }

    @Override
    public Object getResponse() {
        return Boolean.TRUE;
    }

    @Override
    protected void writeInternal(ObjectDataOutput out)
            throws IOException {

        super.writeInternal(out);
        out.writeInt(logicalNodeId);
        out.writeLong(definition.getEpoch().getEpochOffset());
        out.writeInt(definition.getMaxLogicalNodeCount());
        out.writeShort(definition.getBackupCount());
    }

    @Override
    protected void readInternal(ObjectDataInput in)
            throws IOException {

        super.readInternal(in);
        logicalNodeId = in.readInt();

        long epochOffset = in.readLong();
        int maxLogicalNodeCount = in.readInt();
        short backupCount = in.readShort();

        SnowcastEpoch epoch = SnowcastEpoch.byTimestamp(epochOffset);
        definition = new SequencerDefinition(getSequencerName(), epoch, maxLogicalNodeCount, backupCount);
    }

    @Override
    public boolean shouldBackup() {
        return true;
    }

    @Override
    public int getSyncBackupCount() {
        return definition.getBackupCount();
    }

    @Override
    public int getAsyncBackupCount() {
        return 0;
    }

    @Override
    public Operation getBackupOperation() {
        return new BackupDetachLogicalNodeOperation(definition, logicalNodeId, getCallerAddress());
    }
} public class SequencerReplicationOperation
        extends Operation
        implements IdentifiedDataSerializable {

    private PartitionReplication partitionReplication;

    public SequencerReplicationOperation() {
    }

    public SequencerReplicationOperation(PartitionReplication partitionReplication) {
        this.partitionReplication = partitionReplication;
    }

    @Override
    public void run()
            throws Exception {

        NodeSequencerService sequencerService = getService();

        int partitionId = partitionReplication.getPartitionId();
        SequencerPartition partition = sequencerService.getSequencerPartition(partitionId);

        partitionReplication.applyReplication(partition);
    }

    @Override
    public int getFactoryId() {
        return SequencerDataSerializerHook.FACTORY_ID;
    }

    @Override
    public int getId() {
        return SequencerDataSerializerHook.TYPE_REPLICATION_OPERATION;
    }

    @Override
    protected void writeInternal(ObjectDataOutput out)
            throws IOException {

        super.writeInternal(out);
        out.writeObject(partitionReplication);
    }

    @Override
    protected void readInternal(ObjectDataInput in)
            throws IOException {

        super.readInternal(in);
        partitionReplication = in.readObject();
    }
} public interface MessageChannel {

    void sendClientMessage(ClientMessage clientMessage);

    Address getAddress();

    String getUuid();

} public static SnowcastException exception(ExceptionMessages exceptionMessage, Object... exceptionParameters) {
    return exception(SnowcastException::new, exceptionMessage, exceptionParameters);
} public static Supplier<Object[]> exceptionParameters(Object... args) {
    return () -> args;
} public static <V> V execute(ThrowingCallable<V> callable) {
    try {
        return callable.call();
    } catch (Throwable throwable) {
        if (throwable instanceof SnowcastException) {
            throw (SnowcastException) throwable;
        } else if (throwable instanceof Error) {
            throw (Error) throwable;
        }
        throw new SnowcastException(throwable);
    }
} public static final String SERVICE_NAME = "noctarius::SequencerService";
import com.hazelcast.client.impl.protocol.ClientMessage;
import com.hazelcast.client.impl.protocol.codec.SnowcastRegisterChannelCodec;
import com.hazelcast.core.DistributedObject;
import com.hazelcast.instance.BuildInfo;
import com.hazelcast.instance.BuildInfoProvider;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.spi.EventPublishingService;
import com.hazelcast.spi.EventRegistration;
import com.hazelcast.spi.EventService;
import com.hazelcast.spi.InternalCompletableFuture;
import com.hazelcast.spi.InvocationBuilder;
import com.hazelcast.spi.ManagedService;
import com.hazelcast.spi.MigrationAwareService;
import com.hazelcast.spi.NodeEngine;
import com.hazelcast.spi.Operation;
import com.hazelcast.spi.OperationService;
import com.hazelcast.spi.PartitionMigrationEvent;
import com.hazelcast.spi.PartitionReplicationEvent;
import com.hazelcast.spi.RemoteService;
import com.hazelcast.spi.partition.IPartitionService;
import com.hazelcast.spi.partition.MigrationEndpoint;
import com.hazelcast.spi.serialization.SerializationService;
import com.hazelcast.util.Clock;
import com.noctarius.snowcast.SnowcastEpoch;
import com.noctarius.snowcast.SnowcastIllegalStateException;
import com.noctarius.snowcast.SnowcastSequenceState;
import com.noctarius.snowcast.SnowcastSequencer;
import com.noctarius.snowcast.SnowcastSequencerAlreadyRegisteredException;
import com.noctarius.snowcast.impl.operations.AttachLogicalNodeOperation;
import com.noctarius.snowcast.impl.operations.CreateSequencerDefinitionOperation;
import com.noctarius.snowcast.impl.operations.DestroySequencerDefinitionOperation;
import com.noctarius.snowcast.impl.operations.DetachLogicalNodeOperation;
import com.noctarius.snowcast.impl.operations.SequencerReplicationOperation;
import com.noctarius.snowcast.impl.operations.clientcodec.MessageChannel;
import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static com.noctarius.snowcast.impl.ExceptionMessages.INTERNAL_SETUP_FAILED;
import static com.noctarius.snowcast.impl.ExceptionMessages.PARAMETER_IS_NOT_SUPPORTED;
import static com.noctarius.snowcast.impl.ExceptionMessages.SEQUENCER_ALREADY_REGISTERED;
import static com.noctarius.snowcast.impl.ExceptionMessages.UNKNOWN_HAZELCAST_VERSION;
import static com.noctarius.snowcast.impl.ExceptionUtils.exception;
import static com.noctarius.snowcast.impl.ExceptionUtils.exceptionParameters;
import static com.noctarius.snowcast.impl.ExceptionUtils.execute;
import static com.noctarius.snowcast.impl.SnowcastConstants.SERVICE_NAME;
/*
 * Copyright (c) 2015-2017, Christoph Engelbert (aka noctarius) and
 * contributors. 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.noctarius.snowcast.impl;




public class NodeSequencerService
        implements SequencerService, ManagedService, MigrationAwareService, RemoteService,
                   EventPublishingService<Object, Object> {

    private static final MethodType FUTURE_GET_TYPE = MethodType.methodType(Object.class);
    private static final MethodType GET_LISTENER_GET_TYPE = MethodType.methodType(Object.class);

    private final ConcurrentMap<Integer, SequencerPartition> partitions;
    private final ConcurrentMap<String, SequencerProvision> provisions;

    private final MethodHandle getListenerMethodHandle;
    private final MethodHandle futureGetMethodHandle;

    private NodeEngine nodeEngine;
    private EventService eventService;
    private SerializationService serializationService;

    public NodeSequencerService() {
        this.provisions = new ConcurrentHashMap<>();
        this.partitions = new ConcurrentHashMap<>();
        this.getListenerMethodHandle = findEventRegistrationGetListener();
        this.futureGetMethodHandle = findFutureExecutorMethod();
    }

    @Override
    public void init(@Nonnull NodeEngine nodeEngine, @Nullable Properties properties) {
        this.nodeEngine = nodeEngine;
        this.eventService = nodeEngine.getEventService();
        this.serializationService = nodeEngine.getSerializationService();
    }

    @Nonnull
    @Override
    public SnowcastSequencer createSequencer(@Nonnull String sequencerName, @Nonnull SnowcastEpoch epoch,
                                             @Min(128) @Max(8192) int maxLogicalNodeCount, short backupCount) {

        SequencerDefinition definition = new SequencerDefinition(sequencerName, epoch, maxLogicalNodeCount, backupCount);

        Operation operation = new CreateSequencerDefinitionOperation(definition);