2 * Copyright (c) 2016 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.cluster.databroker.actors.dds;
10 import akka.actor.ActorRef;
11 import akka.util.Timeout;
12 import com.google.common.base.Preconditions;
13 import com.google.common.base.Throwables;
14 import com.google.common.primitives.UnsignedLong;
15 import java.util.concurrent.CompletionStage;
16 import java.util.concurrent.TimeUnit;
17 import java.util.concurrent.atomic.AtomicLong;
18 import javax.annotation.Nonnull;
19 import javax.annotation.Nullable;
20 import javax.annotation.concurrent.GuardedBy;
21 import javax.annotation.concurrent.ThreadSafe;
22 import org.opendaylight.controller.cluster.access.ABIVersion;
23 import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
24 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
25 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
26 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
27 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
28 import org.opendaylight.controller.cluster.common.actor.ExplicitAsk;
29 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
32 import scala.Function1;
33 import scala.compat.java8.FutureConverters;
36 * {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
37 * shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
38 * shard leader is resolved via {@link ActorContext}. The product of resolution is {@link ShardBackendInfo}.
40 * @author Robert Varga
43 abstract class AbstractShardBackendResolver extends BackendInfoResolver<ShardBackendInfo> {
44 static final class ShardState {
45 private final CompletionStage<ShardBackendInfo> stage;
47 private ShardBackendInfo result;
49 ShardState(final CompletionStage<ShardBackendInfo> stage) {
50 this.stage = Preconditions.checkNotNull(stage);
51 stage.whenComplete(this::onStageResolved);
54 @Nonnull CompletionStage<ShardBackendInfo> getStage() {
58 @Nullable synchronized ShardBackendInfo getResult() {
62 private synchronized void onStageResolved(final ShardBackendInfo result, final Throwable failure) {
63 if (failure == null) {
64 this.result = Preconditions.checkNotNull(result);
66 LOG.warn("Failed to resolve shard", failure);
71 private static final Logger LOG = LoggerFactory.getLogger(AbstractShardBackendResolver.class);
74 * Fall-over-dead timeout. If we do not make progress in this long, just fall over and propagate the failure.
75 * All users are expected to fail, possibly attempting to recover by restarting. It is fair to remain
78 // TODO: maybe make this configurable somehow?
79 private static final Timeout DEAD_TIMEOUT = Timeout.apply(15, TimeUnit.MINUTES);
81 private final AtomicLong nextSessionId = new AtomicLong();
82 private final Function1<ActorRef, ?> connectFunction;
83 private final ActorContext actorContext;
85 // FIXME: we really need just ActorContext.findPrimaryShardAsync()
86 AbstractShardBackendResolver(final ClientIdentifier clientId, final ActorContext actorContext) {
87 this.actorContext = Preconditions.checkNotNull(actorContext);
88 this.connectFunction = ExplicitAsk.toScala(t -> new ConnectClientRequest(clientId, t, ABIVersion.BORON,
89 ABIVersion.current()));
92 protected final void flushCache(final String shardName) {
93 actorContext.getPrimaryShardInfoCache().remove(shardName);
96 protected final ShardState resolveBackendInfo(final String shardName, final long cookie) {
97 LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
99 return new ShardState(FutureConverters.toJava(actorContext.findPrimaryShardAsync(shardName)).thenCompose(i -> {
100 LOG.debug("Looking up primary info for {} from {}", shardName, i);
101 return FutureConverters.toJava(ExplicitAsk.ask(i.getPrimaryShardActor(), connectFunction, DEAD_TIMEOUT));
102 }).thenApply(response -> {
103 if (response instanceof RequestFailure) {
104 final RequestFailure<?, ?> failure = (RequestFailure<?, ?>) response;
105 LOG.debug("Connect request failed {}", failure, failure.getCause());
106 throw Throwables.propagate(failure.getCause());
109 LOG.debug("Resolved backend information to {}", response);
111 Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response {}", response);
112 final ConnectClientSuccess success = (ConnectClientSuccess) response;
114 return new ShardBackendInfo(success.getBackend(),
115 nextSessionId.getAndIncrement(), success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie),
116 success.getDataTree(), success.getMaxMessages());