package swift.pubsub;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

import swift.clocks.CausalityClock;
import swift.clocks.Timestamp;
import swift.crdt.core.CRDTIdentifier;
import swift.crdt.core.CRDTObjectUpdatesGroup;
import swift.proto.MetadataSamplable;
import swift.proto.MetadataStatsCollector;
import sys.pubsub.PubSub;
import sys.pubsub.PubSub.Notifyable;

/**
 * A notification that brings client's cache from one consistent state to
 * another one.
 * 
 * @author mzawirski
 */
public class BatchUpdatesNotification implements Notifyable<CRDTIdentifier>, MetadataSamplable {
    // TODO: alternatively, for a best-effort FIFO session, include just last
    // id?
    protected CausalityClock oldVersion;
    protected CausalityClock newVersion;
    private boolean newVersionDisasterSafe;
    protected List<CRDTObjectUpdatesGroup<?>> updates;

    BatchUpdatesNotification() {
    }

    public BatchUpdatesNotification(CausalityClock oldVersion, CausalityClock newVersion, boolean disasterSafe,
            List<CRDTObjectUpdatesGroup<?>> updates) {
        this.oldVersion = oldVersion;
        this.newVersion = newVersion;
        this.newVersionDisasterSafe = disasterSafe;
        this.updates = updates;
    }

    @Override
    public void notifyTo(PubSub<CRDTIdentifier> pubsub) {
        ((SwiftSubscriber) pubsub).onNotification(this);
    }

    /**
     * @return old version assumed by the DC
     */
    public CausalityClock getOldVersion() {
        return oldVersion;
    }

    /**
     * @return new consistent version that these packet updates client's cache
     *         to
     */
    public CausalityClock getNewVersion() {
        return newVersion;
    }

    /**
     * @return all updates on client's subscribed object with timestamps between
     *         {@link #getOldVersion()} and {@link #getNewVersion()}
     */
    // FIXME: define subscribed objects with an epochId or so?
    public List<CRDTObjectUpdatesGroup<?>> getUpdates() {
        return updates;
    }

    /**
     * @return true if the version is disaster safe
     */
    public boolean isNewVersionDisasterSafe() {
        return newVersionDisasterSafe;
    }

    @Override
    public Timestamp timestamp() {
        return null;
    }

    @Override
    public Object src() {
        return null;
    }

    @Override
    public CRDTIdentifier key() {
        return null;
    }

    @Override
    public Set<CRDTIdentifier> keys() {
        Set<CRDTIdentifier> keys = new HashSet<CRDTIdentifier>();
        for (final CRDTObjectUpdatesGroup<?> update : updates) {
            keys.add(update.getTargetUID());
        }
        return keys;
    }

    @Override
    public void recordMetadataSample(MetadataStatsCollector collector) {
        // TODO: Marek
    }
}

@@ -33,22 +33,81 @@
 va.util.
-HashSe
+Collections;%0Aimport java.util.HashMap;%0Aimport java.util.LinkedLis
 t;%0Aimpor
@@ -120,24 +120,46 @@
 .util.List;%0A
+import java.util.Map;%0A
 import java.
@@ -520,51 +520,113 @@
 che 
-from one consistent state to%0A * another one
+(a set of subscribed objects) from%0A * the previous consistent state to the new one, described by a vector
 .%0A *
@@ -647,16 +647,20 @@
 zawirski
+,smd
 %0A */%0Apub
@@ -757,138 +757,8 @@
 e %7B%0A
-    // TODO: alternatively, for a best-effort FIFO session, include just last%0A    // id?%0A    protected CausalityClock oldVersion;%0A
     
@@ -851,16 +851,36 @@
 rotected
+ Map%3CCRDTIdentifier,
  List%3CCR
@@ -899,26 +899,34 @@
 tesGroup%3C?%3E%3E
- u
+%3E objectsU
 pdates;%0A%0A   
@@ -1017,35 +1017,8 @@
 ock 
-oldVersion, CausalityClock 
 newV
@@ -1106,46 +1106,8 @@
 ) %7B%0A
-        this.oldVersion = oldVersion;%0A
     
@@ -1209,26 +1209,569 @@
 his.
-updates = updates;
+objectsUpdates = new HashMap%3CCRDTIdentifier, List%3CCRDTObjectUpdatesGroup%3C?%3E%3E%3E();%0A        for (final CRDTObjectUpdatesGroup update : updates) %7B%0A            final CRDTIdentifier id = update.getTargetUID();%0A            List%3CCRDTObjectUpdatesGroup%3C?%3E%3E objectUpdates = objectsUpdates.get(id);%0A            if (objectUpdates == null) %7B%0A                objectUpdates = new LinkedList%3CCRDTObjectUpdatesGroup%3C?%3E%3E();%0A                objectsUpdates.put(id, objectUpdates);%0A            %7D%0A            objectUpdates.add(update.strippedWithCopiedTimestampMappings());%0A        %7D
 %0A   
@@ -1914,147 +1914,8 @@
  %7D%0A%0A
-    /**%0A     * @return old version assumed by the DC%0A     */%0A    public CausalityClock getOldVersion() %7B%0A        return oldVersion;%0A    %7D%0A%0A
     
@@ -2124,16 +2124,25 @@
 @return 
+a map of 
 all upda
@@ -2148,23 +2148,26 @@
 ates on 
-client'
+the object
 s subscr
@@ -2171,22 +2171,45 @@
 scribed 
-object
+by the client,%0A     *        
  with ti
@@ -2224,31 +2224,16 @@
  between
-%0A     *        
  %7B@link 
@@ -2240,11 +2240,11 @@
 #get
-Old
+New
 Vers
@@ -2249,16 +2249,61 @@
 rsion()%7D
+ of the previous%0A     *         notification,
  and %7B@l
@@ -2323,16 +2323,28 @@
 rsion()%7D
+ of this one
 %0A     */
@@ -2417,16 +2417,36 @@
   public
+ Map%3CCRDTIdentifier,
  List%3CCR
@@ -2469,20 +2469,28 @@
 roup%3C?%3E%3E
+%3E
  get
+Objects
 Updates(
@@ -2508,17 +2508,24 @@
  return 
-u
+objectsU
 pdates;%0A
@@ -2980,205 +2980,67 @@
     
-Set%3CCRDTIdentifier%3E keys = new HashSet%3CCRDTIdentifier%3E();%0A        for (final CRDTObjectUpdatesGroup%3C?%3E update : updates) %7B%0A            keys.add(update.getTargetUID());%0A        %7D%0A        return keys
+return Collections.unmodifiableSet(objectsUpdates.keySet())
 ;%0A  

update notifications structure after Sergio's feedback