Change remaining SnapshotSelectionCriteria to prioritize timestamp

[controller.git] / opendaylight / md-sal / sal-remoterpc-connector / src / main / java / org / opendaylight / controller / remote / rpc / registry / gossip / BucketStoreActor.java

diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java
index 6fbe6ceb4d6c1867d17c5f0983aedcb5b51740b3..6c5ca77b037fec1f0ccc6fdc0823fc7a9ae65b8f 100644 (file)
--- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java
+++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
@@ -37,7 +37,7 @@ import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.function.Consumer;
 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
-import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
+import org.opendaylight.controller.remote.rpc.RemoteOpsProviderConfig;
 
 /**
  * A store that syncs its data across nodes in the cluster.
@@ -72,7 +72,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
      */
     private final SetMultimap<ActorRef, Address> watchedActors = HashMultimap.create(1, 1);
 
-    private final RemoteRpcProviderConfig config;
+    private final RemoteOpsProviderConfig config;
     private final String persistenceId;
 
     /**
@@ -88,7 +88,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
     private Integer incarnation;
     private boolean persisting;
 
-    protected BucketStoreActor(final RemoteRpcProviderConfig config, final String persistenceId, final T initialData) {
+    protected BucketStoreActor(final RemoteOpsProviderConfig config, final String persistenceId, final T initialData) {
         this.config = Preconditions.checkNotNull(config);
         this.initialData = Preconditions.checkNotNull(initialData);
         this.persistenceId = Preconditions.checkNotNull(persistenceId);
@@ -106,6 +106,14 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
         return actor -> actor.updateRemoteBuckets(buckets);
     }
 
+    static ExecuteInActor getLocalDataMessage() {
+        return actor -> actor.getSender().tell(actor.getLocalData(), actor.getSelf());
+    }
+
+    static ExecuteInActor getRemoteBucketsMessage() {
+        return actor -> actor.getSender().tell(ImmutableMap.copyOf(actor.getRemoteBuckets()), actor.getSelf());
+    }
+
     public final T getLocalData() {
         return getLocalBucket().getData();
     }
@@ -172,7 +180,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
         } else if (message instanceof SaveSnapshotSuccess) {
             LOG.debug("{}: got command: {}", persistenceId(), message);
             SaveSnapshotSuccess saved = (SaveSnapshotSuccess)message;
-            deleteSnapshots(new SnapshotSelectionCriteria(saved.metadata().sequenceNr(),
+            deleteSnapshots(new SnapshotSelectionCriteria(scala.Long.MaxValue(),
                     saved.metadata().timestamp() - 1, 0L, 0L));
             persisting = false;
             unstash();
@@ -183,7 +191,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
     }
 
     @Override
-    protected final void handleRecover(final Object message) throws Exception {
+    protected final void handleRecover(final Object message) {
         if (message instanceof RecoveryCompleted) {
             if (incarnation != null) {
                 incarnation = incarnation + 1;
@@ -204,7 +212,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
         }
     }
 
-    protected final RemoteRpcProviderConfig getConfig() {
+    protected final RemoteOpsProviderConfig getConfig() {
         return config;
     }
 
@@ -230,14 +238,14 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
      * @param address Remote address
      * @param bucket Bucket removed
      */
-    protected abstract void onBucketRemoved(final Address address, final Bucket<T> bucket);
+    protected abstract void onBucketRemoved(Address address, Bucket<T> bucket);
 
     /**
      * Callback to subclasses invoked when the set of remote buckets is updated.
      *
      * @param newBuckets Map of address to new bucket. Never null, but can be empty.
      */
-    protected abstract void onBucketsUpdated(final Map<Address, Bucket<T>> newBuckets);
+    protected abstract void onBucketsUpdated(Map<Address, Bucket<T>> newBuckets);
 
     /**
      * Helper to collect all known buckets.
@@ -285,6 +293,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
             bucket.getWatchActor().ifPresent(ref -> removeWatch(addr, ref));
             onBucketRemoved(addr, bucket);
         }
+        versions.remove(addr);
     }
 
     /**
@@ -368,7 +377,7 @@ public abstract class BucketStoreActor<T extends BucketData<T>> extends
             versions.remove(addr);
             final Bucket<T> bucket = remoteBuckets.remove(addr);
             if (bucket != null) {
-                LOG.debug("Source actor dead, removing bucket {} from ", bucket, addr);
+                LOG.debug("Source actor dead, removing bucket {} from {}", bucket, addr);
                 onBucketRemoved(addr, bucket);
             }
         }