2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.remote.rpc.registry.gossip;
10 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getBucketsByMembersMessage;
11 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getLocalDataMessage;
12 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.getRemoteBucketsMessage;
13 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.removeBucketMessage;
14 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.updateRemoteBucketsMessage;
16 import akka.actor.ActorRef;
17 import akka.actor.Address;
18 import akka.dispatch.OnComplete;
19 import akka.pattern.Patterns;
20 import akka.util.Timeout;
21 import com.google.common.annotations.Beta;
22 import com.google.common.annotations.VisibleForTesting;
23 import java.util.Collection;
25 import java.util.Objects;
26 import java.util.function.Consumer;
27 import scala.concurrent.ExecutionContext;
28 import scala.concurrent.Future;
31 * Convenience access to {@link BucketStoreActor}. Used mostly by {@link Gossiper}.
33 * @author Robert Varga
37 public final class BucketStoreAccess {
38 private final ActorRef actorRef;
39 private final ExecutionContext dispatcher;
40 private final Timeout timeout;
42 public BucketStoreAccess(final ActorRef actorRef, final ExecutionContext dispatcher, final Timeout timeout) {
43 this.actorRef = Objects.requireNonNull(actorRef);
44 this.dispatcher = Objects.requireNonNull(dispatcher);
45 this.timeout = Objects.requireNonNull(timeout);
48 <T extends BucketData<T>> void getBucketsByMembers(final Collection<Address> members,
49 final Consumer<Map<Address, Bucket<T>>> callback) {
50 Patterns.ask(actorRef, getBucketsByMembersMessage(members), timeout)
51 .onComplete(new OnComplete<Object>() {
52 @SuppressWarnings("unchecked")
54 public void onComplete(final Throwable failure, final Object success) {
55 if (failure == null) {
56 callback.accept((Map<Address, Bucket<T>>) success);
62 void getBucketVersions(final Consumer<Map<Address, Long>> callback) {
63 Patterns.ask(actorRef, Singletons.GET_BUCKET_VERSIONS, timeout).onComplete(new OnComplete<Object>() {
64 @SuppressWarnings("unchecked")
66 public void onComplete(final Throwable failure, final Object success) {
67 if (failure == null) {
68 callback.accept((Map<Address, Long>) success);
74 @SuppressWarnings({ "unchecked", "rawtypes" })
75 public Future<Map<Address, Long>> getBucketVersions() {
76 return (Future) Patterns.ask(actorRef, Singletons.GET_BUCKET_VERSIONS, timeout);
79 @SuppressWarnings("unchecked")
80 void updateRemoteBuckets(final Map<Address, ? extends Bucket<?>> buckets) {
81 actorRef.tell(updateRemoteBucketsMessage((Map<Address, Bucket<?>>) buckets), ActorRef.noSender());
84 void removeRemoteBucket(final Address addr) {
85 actorRef.tell(removeBucketMessage(addr), ActorRef.noSender());
88 @SuppressWarnings({ "unchecked", "rawtypes" })
89 public <T extends BucketData<T>> Future<T> getLocalData() {
90 return (Future) Patterns.ask(actorRef, getLocalDataMessage(), timeout);
93 @SuppressWarnings({ "unchecked", "rawtypes" })
94 public <T extends BucketData<T>> Future<Map<Address, Bucket<T>>> getRemoteBuckets() {
95 return (Future) Patterns.ask(actorRef, getRemoteBucketsMessage(), timeout);
98 public enum Singletons {
99 // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Bucket<?>>
101 // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Long>