X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2Fregistry%2Fgossip%2FBucketStore.java;h=6ffe147e71384b66ece31c4e701294c8630456c8;hb=191ad68d71f30f6ad6258ac458c60c663e5b1b85;hp=3b078aa062fdf1fc3feaca7fa10147de8c3503cc;hpb=9e19dde6d7bbecef432afd4e0576ff18c4a16179;p=controller.git diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java index 3b078aa062..6ffe147e71 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java @@ -12,28 +12,27 @@ import akka.actor.ActorRef; import akka.actor.ActorRefProvider; import akka.actor.Address; import akka.actor.Props; -import akka.actor.UntypedActor; import akka.cluster.ClusterActorRefProvider; import akka.event.Logging; import akka.event.LoggingAdapter; -import org.opendaylight.controller.utils.ConditionalProbe; - import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; - -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetLocalBucket; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetLocalBucketReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateBucket; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets; +import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering; +import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetLocalBucket; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetLocalBucketReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateBucket; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets; +import org.opendaylight.controller.utils.ConditionalProbe; /** * A store that syncs its data across nodes in the cluster. @@ -44,14 +43,14 @@ import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.Bu * This store uses a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}. * */ -public class BucketStore extends UntypedActor { +public class BucketStore extends AbstractUntypedActorWithMetering { final LoggingAdapter log = Logging.getLogger(getContext().system(), this); /** * Bucket owned by the node */ - private BucketImpl localBucket = new BucketImpl();; + private BucketImpl localBucket = new BucketImpl(); /** * Buckets ownded by other known nodes in the cluster @@ -70,27 +69,35 @@ public class BucketStore extends UntypedActor { private ConditionalProbe probe; + private final RemoteRpcProviderConfig config; + + public BucketStore(){ + config = new RemoteRpcProviderConfig(getContext().system().settings().config()); + } + @Override public void preStart(){ ActorRefProvider provider = getContext().provider(); selfAddress = provider.getDefaultAddress(); - if ( provider instanceof ClusterActorRefProvider) - getContext().actorOf(Props.create(Gossiper.class), "gossiper"); + if ( provider instanceof ClusterActorRefProvider) { + getContext().actorOf(Props.create(Gossiper.class).withMailbox(config.getMailBoxName()), "gossiper"); + } } - @Override - public void onReceive(Object message) throws Exception { - - log.debug("Received message: node[{}], message[{}]", selfAddress, message); + @Override + protected void handleReceive(Object message) throws Exception { if (probe != null) { probe.tell(message, getSelf()); } if (message instanceof ConditionalProbe) { + // The ConditionalProbe is only used for unit tests. log.info("Received probe {} {}", getSelf(), message); probe = (ConditionalProbe) message; + // Send back any message to tell the caller we got the probe. + getSender().tell("Got it", getSelf()); } else if (message instanceof UpdateBucket) { receiveUpdateBucket(((UpdateBucket) message).getBucket()); } else if (message instanceof GetAllBuckets) { @@ -99,17 +106,18 @@ public class BucketStore extends UntypedActor { receiveGetLocalBucket(); } else if (message instanceof GetBucketsByMembers) { receiveGetBucketsByMembers( - ((GetBucketsByMembers) message).getMembers()); + ((GetBucketsByMembers) message).getMembers()); } else if (message instanceof GetBucketVersions) { receiveGetBucketVersions(); } else if (message instanceof UpdateRemoteBuckets) { receiveUpdateRemoteBuckets( - ((UpdateRemoteBuckets) message).getBuckets()); + ((UpdateRemoteBuckets) message).getBuckets()); } else { - log.debug("Unhandled message [{}]", message); + if(log.isDebugEnabled()) { + log.debug("Unhandled message [{}]", message); + } unhandled(message); } - } /** @@ -178,13 +186,15 @@ public class BucketStore extends UntypedActor { Map buckets = new HashMap<>(); //first add the local bucket if asked - if (members.contains(selfAddress)) + if (members.contains(selfAddress)) { buckets.put(selfAddress, localBucket); + } //then get buckets for requested remote nodes for (Address address : members){ - if (remoteBuckets.containsKey(address)) + if (remoteBuckets.containsKey(address)) { buckets.put(address, remoteBuckets.get(address)); + } } return buckets; @@ -208,7 +218,9 @@ public class BucketStore extends UntypedActor { void receiveUpdateRemoteBuckets(Map receivedBuckets){ if (receivedBuckets == null || receivedBuckets.isEmpty()) + { return; //nothing to do + } //Remote cant update self's bucket receivedBuckets.remove(selfAddress); @@ -216,15 +228,20 @@ public class BucketStore extends UntypedActor { for (Map.Entry entry : receivedBuckets.entrySet()){ Long localVersion = versions.get(entry.getKey()); - if (localVersion == null) localVersion = -1L; + if (localVersion == null) { + localVersion = -1L; + } Bucket receivedBucket = entry.getValue(); - if (receivedBucket == null) + if (receivedBucket == null) { continue; + } Long remoteVersion = receivedBucket.getVersion(); - if (remoteVersion == null) remoteVersion = -1L; + if (remoteVersion == null) { + remoteVersion = -1L; + } //update only if remote version is newer if ( remoteVersion.longValue() > localVersion.longValue() ) { @@ -232,8 +249,9 @@ public class BucketStore extends UntypedActor { versions.put(entry.getKey(), remoteVersion); } } - - log.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets); + if(log.isDebugEnabled()) { + log.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets); + } } ///