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.removeBucketMessage;
12 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreActor.updateRemoteBucketsMessage;
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;
25 import java.util.function.Consumer;
28 * Convenience access to {@link BucketStoreActor}. Used mostly by {@link Gossiper}.
30 * @author Robert Varga
34 public final class BucketStoreAccess {
35 private final ActorContext context;
36 private final Timeout timeout;
38 BucketStoreAccess(final ActorContext context, final Timeout timeout) {
39 this.context = Preconditions.checkNotNull(context);
40 this.timeout = Preconditions.checkNotNull(timeout);
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")
49 public void onComplete(final Throwable failure, final Object success) {
50 if (failure == null) {
51 callback.accept((Map<Address, Bucket<T>>) success);
54 }, context.dispatcher());
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")
61 public void onComplete(final Throwable failure, final Object success) {
62 if (failure == null) {
63 callback.accept((Map<Address, Long>) success);
66 }, context.dispatcher());
69 @SuppressWarnings("unchecked")
70 void updateRemoteBuckets(final Map<Address, ? extends Bucket<?>> buckets) {
71 context.parent().tell(updateRemoteBucketsMessage((Map<Address, Bucket<?>>) buckets), ActorRef.noSender());
74 void removeRemoteBucket(final Address addr) {
75 context.parent().tell(removeBucketMessage(addr), ActorRef.noSender());
78 public enum Singletons {
79 // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Bucket<?>>
81 // Sent from Gossiper to BucketStore, response is an immutable Map<Address, Long>