package swift.crdt;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import swift.clocks.CausalityClock;
import swift.clocks.CausalityClock.CMP_CLOCK;
import swift.clocks.Timestamp;
import swift.clocks.TripleTimestamp;
import swift.crdt.interfaces.CRDTUpdate;
import swift.crdt.interfaces.Copyable;
import swift.crdt.interfaces.TxnHandle;
import swift.crdt.interfaces.TxnLocalCRDT;

public class RegisterVersioned<V extends Copyable> extends BaseCRDT<RegisterVersioned<V>> {
    private static final long serialVersionUID = 1L;

    // Queue holding the versioning information, ordering is compatible with
    // causal dependency, newest entries coming first
    private SortedSet<QueueEntry<V>> values;

    public RegisterVersioned() {
        this.values = new TreeSet<QueueEntry<V>>();
    }

    @Override
    public void rollback(Timestamp ts) {
        Iterator<QueueEntry<V>> it = values.iterator();
        while (it.hasNext()) {
            QueueEntry<V> entry = it.next();
            if (!entry.c.includes(ts)) {
                break;
            }
            if (entry.ts.equals(ts)) {
                it.remove();
            }
        }
    }

    @Override
    protected void pruneImpl(CausalityClock pruningPoint) {
        // at least one value should remain after pruning
        if (values.size() == 1) {
            return;
        }
        // remove all versions older than the pruningPoint
        SortedSet<QueueEntry<V>> pruned = new TreeSet<QueueEntry<V>>();
        for (QueueEntry<V> e : values) {
            if (!pruningPoint.includes(e.ts)) {
                pruned.add(e);
            }
        }
        if (pruned.isEmpty()) {
            pruned = new TreeSet<QueueEntry<V>>();
            pruned.add(values.first());
        }
        values = pruned;
    }

    public void update(V val, TripleTimestamp ts, CausalityClock c) {
        values.add(new QueueEntry<V>(ts, c, val));
    }

    @Override
    protected void mergePayload(RegisterVersioned<V> otherObject) {
        CMP_CLOCK cmpClock = otherObject.getPruneClock().compareTo(getPruneClock());
        if (cmpClock == CMP_CLOCK.CMP_DOMINATES) {
            pruneImpl(otherObject.getPruneClock());
        } else {
            if (cmpClock == CMP_CLOCK.CMP_ISDOMINATED) {
                otherObject.pruneImpl(getPruneClock());
            }
        }
        values.addAll(otherObject.values);
    }

    @Override
    protected TxnLocalCRDT<RegisterVersioned<V>> getTxnLocalCopyImpl(CausalityClock versionClock, TxnHandle txn) {
        final RegisterVersioned<V> creationState = isRegisteredInStore() ? null : new RegisterVersioned<V>();
        RegisterTxnLocal<V> localview = new RegisterTxnLocal<V>(id, txn, versionClock, creationState,
                value(versionClock));
        return localview;
    }

    private V value(CausalityClock versionClock) {
        for (QueueEntry<V> e : values) {
            if (versionClock.includes(e.ts)) {
                return e.value;
            }
        }
        return null;
    }

    @Override
    public String toString() {
        return values.toString();
    }

    @Override
    protected void execute(CRDTUpdate<RegisterVersioned<V>> op) {
        op.applyTo(this);
    }

    @Override
    public RegisterVersioned<V> copy() {
        RegisterVersioned<V> copyObj = new RegisterVersioned<V>();
        for (QueueEntry<V> e : values) {
            copyObj.values.add(e.copy());
        }
        copyBase(copyObj);
        return copyObj;
    }

    @Override
    protected Set<Timestamp> getUpdateTimestampsSinceImpl(CausalityClock clock) {
        final Set<Timestamp> result = new HashSet<Timestamp>();
        for (QueueEntry<V> e : values) {
            if (!clock.includes(e.ts)) {
                result.add(e.ts.cloneBaseTimestamp());
            }
        }
        return result;
    }

    // public for sake of Kryo...
    public static class QueueEntry<V extends Copyable> implements Comparable<QueueEntry<V>>, Serializable {
        TripleTimestamp ts;
        CausalityClock c;
        V value;

        /**
         * Do not use: Empty constructor only to be used by Kryo serialization.
         */
        public QueueEntry() {
        }

        public QueueEntry(TripleTimestamp ts, CausalityClock c, V value) {
            this.ts = ts;
            this.c = c;
            this.value = value;
        }

        @Override
        public int compareTo(QueueEntry<V> other) {
            CMP_CLOCK result = this.c.compareTo(other.c);
            switch (result) {
            case CMP_CONCURRENT:
            case CMP_EQUALS:
                if (other.ts == null) {
                    return 1;
                } else {
                    return other.ts.compareTo(this.ts);
                }
            case CMP_ISDOMINATED:
                return 1;
            case CMP_DOMINATES:
                return -1;
            default:
                return 0;
            }
        }

        @Override
        public String toString() {
            return value + " -> " + ts + "," + c;
        }

        public QueueEntry<V> copy() {
            QueueEntry<V> copyObj = new QueueEntry<V>(ts, c.clone(), (V) value.copy());
            return copyObj;
        }
    }
}

@@ -1355,54 +1355,72 @@
  // 
-at least one value should remain after pruning
+Remove all values older than the pruningPoint, except the single
 %0A   
@@ -1428,232 +1428,233 @@
     
-if (
+// 
 value
-s.size() == 1) %7B%0A            return;%0A        %7D%0A        // remove all versions older than the pruningPoint%0A        SortedSet%3CQueueEntry%3CV%3E%3E pruned = new TreeSet%3CQueueEntry%3CV%3E%3E();%0A        for (QueueEntry%3CV%3E e : values
+ representing purningPoint - there must be a summary of pruned%0A        // state.%0A        boolean firstMatchSkipped = false;%0A        final Iterator%3CQueueEntry%3CV%3E%3E iter = values.iterator();%0A        while (iter.hasNext()
 ) %7B%0A
@@ -1669,17 +1669,16 @@
     if (
-!
 pruningP
@@ -1683,33 +1683,43 @@
 gPoint.includes(
-e
+iter.next()
 .ts)) %7B%0A        
@@ -1730,22 +1730,32 @@
     
-pruned.add(e);
+if (firstMatchSkipped) %7B
 %0A   
@@ -1759,34 +1759,32 @@
             
-%7D%0A
         
 %7D%0A        if
@@ -1775,167 +1775,135 @@
     
-%7D%0A        if (pruned.isEmpty()) %7B%0A            pruned = new TreeSet%3CQueueEntry%3CV%3E%3E();%0A            pruned.add(values.first());%0A        %7D%0A        values = pruned;
+iter.remove();%0A                %7D else %7B%0A                    firstMatchSkipped = true;%0A                %7D%0A            %7D%0A        %7D
 %0A   

fix pruneImpl() to summarize the pruned state by a single entry