X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2Fregistry%2Fgossip%2FBucketStore.java;h=6ffe147e71384b66ece31c4e701294c8630456c8;hp=2f634ce1fabffea20397cd21cf6196c982418c2d;hb=7f2ecd54dd3eb09af469a610fdd541b48ed95b80;hpb=d7ce7c5acee7e6f7cd7895ceff5af63ac53789ad 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 2f634ce1fa..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 @@ -9,29 +9,30 @@ package org.opendaylight.controller.remote.rpc.registry.gossip; import akka.actor.ActorRef; +import akka.actor.ActorRefProvider; import akka.actor.Address; import akka.actor.Props; -import akka.actor.UntypedActor; -import akka.cluster.Cluster; +import akka.cluster.ClusterActorRefProvider; import akka.event.Logging; import akka.event.LoggingAdapter; - 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. @@ -42,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 @@ -64,55 +65,59 @@ public class BucketStore extends UntypedActor { /** * Cluster address for this node */ - private final Address selfAddress = Cluster.get(getContext().system()).selfAddress(); + private Address selfAddress; - /** - * Our private gossiper - */ - private ActorRef gossiper; + private ConditionalProbe probe; + + private final RemoteRpcProviderConfig config; public BucketStore(){ - gossiper = getContext().actorOf(Props.create(Gossiper.class), "gossiper"); + config = new RemoteRpcProviderConfig(getContext().system().settings().config()); } - /** - * This constructor is useful for testing. - * TODO: Pass Props instead of ActorRef - * - * @param gossiper - */ - public BucketStore(ActorRef gossiper){ - this.gossiper = gossiper; + @Override + public void preStart(){ + ActorRefProvider provider = getContext().provider(); + selfAddress = provider.getDefaultAddress(); + + 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 UpdateBucket) + 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) + } else if (message instanceof GetAllBuckets) { receiveGetAllBucket(); - - else if (message instanceof GetLocalBucket) + } else if (message instanceof GetLocalBucket) { receiveGetLocalBucket(); - - else if (message instanceof GetBucketsByMembers) - receiveGetBucketsByMembers(((GetBucketsByMembers) message).getMembers()); - - else if (message instanceof GetBucketVersions) + } else if (message instanceof GetBucketsByMembers) { + receiveGetBucketsByMembers( + ((GetBucketsByMembers) message).getMembers()); + } else if (message instanceof GetBucketVersions) { receiveGetBucketVersions(); - - else if (message instanceof UpdateRemoteBuckets) - receiveUpdateRemoteBuckets(((UpdateRemoteBuckets) message).getBuckets()); - - else { - log.debug("Unhandled message [{}]", message); + } else if (message instanceof UpdateRemoteBuckets) { + receiveUpdateRemoteBuckets( + ((UpdateRemoteBuckets) message).getBuckets()); + } else { + if(log.isDebugEnabled()) { + log.debug("Unhandled message [{}]", message); + } unhandled(message); } - } /** @@ -181,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; @@ -211,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); @@ -219,24 +228,30 @@ 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 > localVersion ) { + if ( remoteVersion.longValue() > localVersion.longValue() ) { remoteBuckets.put(entry.getKey(), receivedBucket); 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); + } } ///