package com.atlassian.stash.internal.concurrent;

import com.atlassian.hazelcast.BaseEntryProcessor;
import com.atlassian.hazelcast.serialization.OsgiSafe;
import com.google.common.collect.Lists;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.DataSerializable;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;

/**
 * EntryProcessor for {@link HazelcastBucketedExecutor} that is submitted to the IMap when processing of a bucket of
 * tasks finishes. If processing succeeded, the 'executing' flag on the TaskBucket is simply cleared. However, if
 * processing failed, all previously {@link ClaimTasksFromBucketProcessor claimed} tasks are returned to the head of
 * the {@link TaskBucket}.
 */
class FinishProcessingBucketProcessor<T> extends BaseEntryProcessor<BucketKey, TaskBucket<OsgiSafe<T>>>
        implements DataSerializable {

    private Collection<OsgiSafe<T>> returned;

    @SuppressWarnings("UnusedDeclaration")
    FinishProcessingBucketProcessor() {
    }

    FinishProcessingBucketProcessor(Collection<OsgiSafe<T>> returned) {
        this.returned = returned;
    }

    @Override
    public Object process(Map.Entry<BucketKey, TaskBucket<OsgiSafe<T>>> entry) {
        TaskBucket<OsgiSafe<T>> bucket = entry.getValue();
        if (bucket == null) {
            if (returned == null || returned.isEmpty()) {
                return null;
            }
            bucket = new TaskBucket<>();
        }

        if (returned != null && !returned.isEmpty()) {
            bucket.returnAll(returned);
        }
        bucket.setExecuting(false);
        entry.setValue(bucket.isEmpty() ? null : bucket);
        return null;
    }

    @Override
    public void readData(ObjectDataInput in) throws IOException {
        int size = in.readInt();
        returned = Lists.newArrayListWithCapacity(size);
        for (int i = 0; i < size; ++i) {
            returned.add(in.readObject());
        }
    }

    @Override
    public void writeData(ObjectDataOutput out) throws IOException {
        if (returned == null || returned.isEmpty()) {
            out.writeInt(0);
        } else {
            out.writeInt(returned.size());
            for (OsgiSafe<T> value : returned) {
                out.writeObject(value);
            }
        }
    }
}
