opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreAccess.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.remote.rpc.registry.gossip;
   9
  10 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getBucketsByMembersMessage;
  11 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.removeBucketMessage;
  12 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.updateRemoteBucketsMessage;
  13
  14 import akka.actor.ActorContext;
  15 import akka.actor.ActorRef;
  16 import akka.actor.Address;
  17 import akka.dispatch.OnComplete;
  18 import akka.pattern.Patterns;
  19 import akka.util.Timeout;
  20 import com.google.common.annotations.Beta;
  21 import com.google.common.annotations.VisibleForTesting;
  22 import com.google.common.base.Preconditions;
  23 import java.util.Collection;
  24 import java.util.Map;
  25 import java.util.function.Consumer;
  26
  27 /**
  28  * Convenience access to {@link BucketStoreActor}. Used mostly by {@link Gossiper}.
  29  *
  30  * @author Robert Varga
  31  */
  32 @Beta
  33 @VisibleForTesting
  34 public final class BucketStoreAccess {
  35     private final ActorContext context;
  36     private final Timeout timeout;
  37
  38     BucketStoreAccess(final ActorContext context, final Timeout timeout) {
  39         this.context = Preconditions.checkNotNull(context);
  40         this.timeout = Preconditions.checkNotNull(timeout);
  41     }
  42
  43     <T extends BucketData<T>> void getBucketsByMembers(final Collection<Address> members,
  44             final Consumer<Map<Address, Bucket<T>>> callback) {
  45         Patterns.ask(context.parent(), getBucketsByMembersMessage(members), timeout)
  46             .onComplete(new OnComplete<Object>() {
  47                 @SuppressWarnings("unchecked")
  48                 @Override
  49                 public void onComplete(final Throwable failure, final Object success) {
  50                     if (failure == null) {
  51                         callback.accept((Map<Address, Bucket<T>>) success);
  52                     }
  53                 }
  54             }, context.dispatcher());
  55     }
  56
  57     void getBucketVersions(final Consumer<Map<Address, Long>> callback) {
  58         Patterns.ask(context.parent(), Singletons.GET_BUCKET_VERSIONS, timeout).onComplete(new OnComplete<Object>() {
  59             @SuppressWarnings("unchecked")
  60             @Override
  61             public void onComplete(final Throwable failure, final Object success) {
  62                 if (failure == null) {
  63                     callback.accept((Map<Address, Long>) success);
  64                 }
  65             }
  66         }, context.dispatcher());
  67     }
  68
  69     @SuppressWarnings("unchecked")
  70     void updateRemoteBuckets(final Map<Address, ? extends Bucket<?>> buckets) {
  71         context.parent().tell(updateRemoteBucketsMessage((Map<Address, Bucket<?>>) buckets), ActorRef.noSender());
  72     }
  73
  74     void removeRemoteBucket(final Address addr) {
  75         context.parent().tell(removeBucketMessage(addr), ActorRef.noSender());
  76     }
  77
  78     public enum Singletons {
  79         // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Bucket<?>>
  80         GET_ALL_BUCKETS,
  81         // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Long>
  82         GET_BUCKET_VERSIONS,
  83     }
  84 }