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 java.util.Objects.requireNonNull;
  11 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getBucketsByMembersMessage;
  12 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getLocalDataMessage;
  13 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getRemoteBucketsMessage;
  14 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.removeBucketMessage;
  15 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.updateRemoteBucketsMessage;
  16
  17 import akka.actor.ActorRef;
  18 import akka.actor.Address;
  19 import akka.dispatch.OnComplete;
  20 import akka.pattern.Patterns;
  21 import akka.util.Timeout;
  22 import com.google.common.annotations.VisibleForTesting;
  23 import java.util.Collection;
  24 import java.util.Map;
  25 import java.util.function.Consumer;
  26 import scala.concurrent.ExecutionContext;
  27 import scala.concurrent.Future;
  28
  29 /**
  30  * Convenience access to {@link BucketStoreActor}. Used mostly by {@link Gossiper}.
  31  */
  32 @VisibleForTesting
  33 public final class BucketStoreAccess {
  34     private final ActorRef actorRef;
  35     private final ExecutionContext dispatcher;
  36     private final Timeout timeout;
  37
  38     public BucketStoreAccess(final ActorRef actorRef, final ExecutionContext dispatcher, final Timeout timeout) {
  39         this.actorRef = requireNonNull(actorRef);
  40         this.dispatcher = requireNonNull(dispatcher);
  41         this.timeout = requireNonNull(timeout);
  42     }
  43
  44     <T extends BucketData<T>> void getBucketsByMembers(final Collection<Address> members,
  45             final Consumer<Map<Address, Bucket<T>>> callback) {
  46         Patterns.ask(actorRef, getBucketsByMembersMessage(members), timeout)
  47             .onComplete(new OnComplete<>() {
  48                 @SuppressWarnings("unchecked")
  49                 @Override
  50                 public void onComplete(final Throwable failure, final Object success) {
  51                     if (failure == null) {
  52                         callback.accept((Map<Address, Bucket<T>>) success);
  53                     }
  54                 }
  55             }, dispatcher);
  56     }
  57
  58     void getBucketVersions(final Consumer<Map<Address, Long>> callback) {
  59         Patterns.ask(actorRef, Singletons.GET_BUCKET_VERSIONS, timeout).onComplete(new OnComplete<>() {
  60             @SuppressWarnings("unchecked")
  61             @Override
  62             public void onComplete(final Throwable failure, final Object success) {
  63                 if (failure == null) {
  64                     callback.accept((Map<Address, Long>) success);
  65                 }
  66             }
  67         }, dispatcher);
  68     }
  69
  70     @SuppressWarnings({ "unchecked", "rawtypes" })
  71     public Future<Map<Address, Long>> getBucketVersions() {
  72         return (Future) Patterns.ask(actorRef, Singletons.GET_BUCKET_VERSIONS, timeout);
  73     }
  74
  75     @SuppressWarnings("unchecked")
  76     void updateRemoteBuckets(final Map<Address, ? extends Bucket<?>> buckets) {
  77         actorRef.tell(updateRemoteBucketsMessage((Map<Address, Bucket<?>>) buckets), ActorRef.noSender());
  78     }
  79
  80     void removeRemoteBucket(final Address addr) {
  81         actorRef.tell(removeBucketMessage(addr), ActorRef.noSender());
  82     }
  83
  84     @SuppressWarnings({ "unchecked", "rawtypes" })
  85     public <T extends BucketData<T>> Future<T> getLocalData() {
  86         return (Future) Patterns.ask(actorRef, getLocalDataMessage(), timeout);
  87     }
  88
  89     @SuppressWarnings({ "unchecked", "rawtypes" })
  90     public <T extends BucketData<T>> Future<Map<Address, Bucket<T>>> getRemoteBuckets() {
  91         return (Future) Patterns.ask(actorRef, getRemoteBucketsMessage(), timeout);
  92     }
  93
  94     public enum Singletons {
  95         /**
  96          * Sent from Gossiper to BucketStore, response is an immutable {@code Map&lt;Address, Bucket&lt;?&gt;&gt;}.
  97          */
  98         GET_ALL_BUCKETS,
  99         /**
 100          * Sent from Gossiper to BucketStore, response is an immutable {@code Map&lt;Address, Long&gt;}.
 101          */
 102         GET_BUCKET_VERSIONS,
 103     }
 104 }