package com.atlassian.stash.internal.concurrent;

import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.DataSerializable;

import java.io.IOException;
import java.util.*;

/**
 * Represents a 'bucket' of tasks used by {@link HazelcastBucketedExecutor}. A {@code TaskBucket} contains a list of
 * tasks and an 'executing' flag.
 *
 * @param <T> the task type
 */
class TaskBucket<T> implements DataSerializable {
    private LinkedHashSet<T> scheduled;
    private boolean executing;

    TaskBucket() {
        scheduled = new LinkedHashSet<>();
    }

    boolean add(T task) {
        return scheduled.add(task);
    }

    boolean isEmpty() {
        return scheduled.isEmpty();
    }

    boolean isExecuting() {
        return executing;
    }

    @Override
    public void readData(ObjectDataInput objectDataInput) throws IOException {
        scheduled = new LinkedHashSet<>();
        int items = objectDataInput.readInt();
        for (int i = 0; i < items; ++i) {
            scheduled.add(objectDataInput.readObject());
        }
        executing = objectDataInput.readBoolean();
    }

    void returnAll(Collection<T> tasks) {
        LinkedHashSet<T> combined = new LinkedHashSet<>(tasks);
        combined.addAll(scheduled);
        scheduled = combined;
    }

    void setExecuting(boolean executing) {
        this.executing = executing;
    }

    // NOTE: this method returns an ArrayList because we care about the order of items and because Hazelcast does
    // not have a serializer for LinkedHashSet. If we'd return a LinkedHashSet, we'd get a NotSerializableException
    // (from the return value of ClaimTasksFromBucketProcessor) because OsgiSafe does not implement Serializable.
    List<T> take(int max) {
        if (max <= 0 || scheduled.isEmpty()) {
            return Collections.emptyList();
        }

        if (max >= scheduled.size()) {
            List<T> result = new ArrayList<>(scheduled);
            scheduled = new LinkedHashSet<>();
            return result;
        }

        List<T> result = new ArrayList<>(max);
        Iterator<T> it = scheduled.iterator();
        for (int i = 0; i < max; ++i) {
            result.add(it.next());
            it.remove();
        }

        return result;
    }

    @Override
    public void writeData(ObjectDataOutput objectDataOutput) throws IOException {
        objectDataOutput.writeInt(scheduled.size());
        for (T task : scheduled) {
            objectDataOutput.writeObject(task);
        }
        objectDataOutput.writeBoolean(executing);
    }
}
