opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java

   1 /*
   2  * Copyright (c) 2017 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.clustering.it.provider;
   9
  10 import akka.actor.ActorRef;
  11 import akka.dispatch.Futures;
  12 import akka.dispatch.OnComplete;
  13 import akka.pattern.Patterns;
  14 import com.google.common.base.Strings;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.SettableFuture;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.Optional;
  20 import java.util.concurrent.ExecutionException;
  21 import java.util.concurrent.TimeUnit;
  22 import java.util.concurrent.TimeoutException;
  23 import org.opendaylight.controller.cluster.ActorSystemProvider;
  24 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  25 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  26 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
  27 import org.opendaylight.controller.clustering.it.provider.impl.FlappingSingletonService;
  28 import org.opendaylight.controller.clustering.it.provider.impl.GetConstantService;
  29 import org.opendaylight.controller.clustering.it.provider.impl.IdIntsListener;
  30 import org.opendaylight.controller.clustering.it.provider.impl.PublishNotificationsTask;
  31 import org.opendaylight.controller.clustering.it.provider.impl.RoutedGetConstantService;
  32 import org.opendaylight.controller.clustering.it.provider.impl.SingletonGetConstantService;
  33 import org.opendaylight.controller.clustering.it.provider.impl.WriteTransactionsHandler;
  34 import org.opendaylight.controller.clustering.it.provider.impl.YnlListener;
  35 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
  36 import org.opendaylight.mdsal.binding.api.NotificationService;
  37 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  38 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  39 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  40 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  41 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  42 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
  43 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  44 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  45 import org.opendaylight.mdsal.dom.api.DOMRpcImplementationRegistration;
  46 import org.opendaylight.mdsal.dom.api.DOMRpcProviderService;
  47 import org.opendaylight.mdsal.dom.api.DOMSchemaService;
  48 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  49 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  50 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.AddShardReplicaInput;
  51 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.AddShardReplicaOutput;
  52 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsInput;
  53 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsOutput;
  54 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsOutputBuilder;
  55 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.IsClientAbortedInput;
  56 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.IsClientAbortedOutput;
  57 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.OdlMdsalLowlevelControlService;
  58 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterBoundConstantInput;
  59 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterBoundConstantOutput;
  60 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterBoundConstantOutputBuilder;
  61 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterConstantInput;
  62 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterConstantOutput;
  63 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterConstantOutputBuilder;
  64 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterDefaultConstantInput;
  65 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterDefaultConstantOutput;
  66 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterFlappingSingletonInput;
  67 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterFlappingSingletonOutput;
  68 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterFlappingSingletonOutputBuilder;
  69 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterSingletonConstantInput;
  70 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterSingletonConstantOutput;
  71 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterSingletonConstantOutputBuilder;
  72 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RemoveShardReplicaInput;
  73 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RemoveShardReplicaOutput;
  74 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ShutdownShardReplicaInput;
  75 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ShutdownShardReplicaOutput;
  76 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ShutdownShardReplicaOutputBuilder;
  77 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.StartPublishNotificationsInput;
  78 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.StartPublishNotificationsOutput;
  79 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.StartPublishNotificationsOutputBuilder;
  80 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeDdtlInput;
  81 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeDdtlOutput;
  82 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeDtclInput;
  83 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeDtclOutput;
  84 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeDtclOutputBuilder;
  85 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeYnlInput;
  86 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeYnlOutput;
  87 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.SubscribeYnlOutputBuilder;
  88 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterBoundConstantInput;
  89 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterBoundConstantOutput;
  90 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterBoundConstantOutputBuilder;
  91 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterConstantInput;
  92 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterConstantOutput;
  93 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterConstantOutputBuilder;
  94 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterDefaultConstantInput;
  95 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterDefaultConstantOutput;
  96 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterFlappingSingletonInput;
  97 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterFlappingSingletonOutput;
  98 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterFlappingSingletonOutputBuilder;
  99 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterSingletonConstantInput;
 100 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterSingletonConstantOutput;
 101 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterSingletonConstantOutputBuilder;
 102 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDdtlInput;
 103 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDdtlOutput;
 104 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDtclInput;
 105 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDtclOutput;
 106 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDtclOutputBuilder;
 107 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlInput;
 108 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlOutput;
 109 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsInput;
 110 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutput;
 111 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 112 import org.opendaylight.yangtools.concepts.ObjectRegistration;
 113 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 114 import org.opendaylight.yangtools.yang.common.ErrorTag;
 115 import org.opendaylight.yangtools.yang.common.ErrorType;
 116 import org.opendaylight.yangtools.yang.common.RpcResult;
 117 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 118 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 119 import org.slf4j.Logger;
 120 import org.slf4j.LoggerFactory;
 121 import scala.concurrent.duration.FiniteDuration;
 122
 123 public final class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService {
 124     private static final Logger LOG = LoggerFactory.getLogger(MdsalLowLevelTestProvider.class);
 125
 126     private final ObjectRegistration<OdlMdsalLowlevelControlService> registration;
 127     private final DistributedDataStoreInterface configDataStore;
 128     private final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer;
 129     private final DOMDataBroker domDataBroker;
 130     private final NotificationPublishService notificationPublishService;
 131     private final NotificationService notificationService;
 132     private final ClusterSingletonServiceProvider singletonService;
 133     private final DOMRpcProviderService domRpcService;
 134     private final DOMDataTreeChangeService domDataTreeChangeService;
 135
 136     private final Map<InstanceIdentifier<?>, DOMRpcImplementationRegistration<RoutedGetConstantService>>
 137             routedRegistrations = new HashMap<>();
 138
 139     private final Map<String, ListenerRegistration<YnlListener>> ynlRegistrations = new HashMap<>();
 140
 141     private DOMRpcImplementationRegistration<GetConstantService> globalGetConstantRegistration = null;
 142     private ClusterSingletonServiceRegistration getSingletonConstantRegistration;
 143     private FlappingSingletonService flappingSingletonService;
 144     private ListenerRegistration<DOMDataTreeChangeListener> dtclReg;
 145     private IdIntsListener idIntsListener;
 146     private final Map<String, PublishNotificationsTask> publishNotificationsTasks = new HashMap<>();
 147
 148     public MdsalLowLevelTestProvider(
 149                                      // FIXME: do not depend on this service
 150                                      final RpcProviderService rpcRegistry,
 151                                      final DOMRpcProviderService domRpcService,
 152                                      final ClusterSingletonServiceProvider singletonService,
 153                                      // FIXME: do not depend on this service
 154                                      final DOMSchemaService schemaService,
 155                                      final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer,
 156                                      final NotificationPublishService notificationPublishService,
 157                                      final NotificationService notificationService,
 158                                      final DOMDataBroker domDataBroker,
 159                                      final DistributedDataStoreInterface configDataStore,
 160                                      // FIXME: do not depend on this service
 161                                      final ActorSystemProvider actorSystemProvider) {
 162         this.domRpcService = domRpcService;
 163         this.singletonService = singletonService;
 164         this.bindingNormalizedNodeSerializer = bindingNormalizedNodeSerializer;
 165         this.notificationPublishService = notificationPublishService;
 166         this.notificationService = notificationService;
 167         this.domDataBroker = domDataBroker;
 168         this.configDataStore = configDataStore;
 169
 170         domDataTreeChangeService = domDataBroker.getExtensions().getInstance(DOMDataTreeChangeService.class);
 171
 172         registration = rpcRegistry.registerRpcImplementation(OdlMdsalLowlevelControlService.class, this);
 173     }
 174
 175     @Override
 176     @SuppressWarnings("checkstyle:IllegalCatch")
 177     public ListenableFuture<RpcResult<UnregisterSingletonConstantOutput>> unregisterSingletonConstant(
 178             final UnregisterSingletonConstantInput input) {
 179         LOG.info("In unregisterSingletonConstant");
 180
 181         if (getSingletonConstantRegistration == null) {
 182             return RpcResultBuilder.<UnregisterSingletonConstantOutput>failed()
 183                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING, "No prior RPC was registered")
 184                 .buildFuture();
 185         }
 186
 187         try {
 188             getSingletonConstantRegistration.close();
 189             getSingletonConstantRegistration = null;
 190
 191             return RpcResultBuilder.success(new UnregisterSingletonConstantOutputBuilder().build()).buildFuture();
 192         } catch (Exception e) {
 193             String msg = "Error closing the singleton constant service";
 194             LOG.error(msg, e);
 195             return RpcResultBuilder.<UnregisterSingletonConstantOutput>failed().withError(
 196                     ErrorType.APPLICATION, msg, e).buildFuture();
 197         }
 198     }
 199
 200     @Override
 201     public ListenableFuture<RpcResult<StartPublishNotificationsOutput>> startPublishNotifications(
 202             final StartPublishNotificationsInput input) {
 203         LOG.info("In startPublishNotifications - input: {}", input);
 204
 205         final PublishNotificationsTask task = new PublishNotificationsTask(notificationPublishService, input.getId(),
 206                 input.getSeconds().toJava(), input.getNotificationsPerSecond().toJava());
 207
 208         publishNotificationsTasks.put(input.getId(), task);
 209
 210         task.start();
 211
 212         return RpcResultBuilder.success(new StartPublishNotificationsOutputBuilder().build()).buildFuture();
 213     }
 214
 215     @Override
 216     public ListenableFuture<RpcResult<SubscribeDtclOutput>> subscribeDtcl(final SubscribeDtclInput input) {
 217         LOG.info("In subscribeDtcl - input: {}", input);
 218
 219         if (dtclReg != null) {
 220             return RpcResultBuilder.<SubscribeDtclOutput>failed().withError(ErrorType.RPC, ErrorTag.DATA_EXISTS,
 221                 "There is already a DataTreeChangeListener registered for id-ints")
 222                 .buildFuture();
 223         }
 224
 225         idIntsListener = new IdIntsListener();
 226
 227         dtclReg = domDataTreeChangeService.registerDataTreeChangeListener(
 228             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, WriteTransactionsHandler.ID_INT_YID),
 229             idIntsListener);
 230
 231         return RpcResultBuilder.success(new SubscribeDtclOutputBuilder().build()).buildFuture();
 232     }
 233
 234     @Override
 235     public ListenableFuture<RpcResult<WriteTransactionsOutput>> writeTransactions(final WriteTransactionsInput input) {
 236         return WriteTransactionsHandler.start(domDataBroker, input);
 237     }
 238
 239     @Override
 240     public ListenableFuture<RpcResult<IsClientAbortedOutput>> isClientAborted(final IsClientAbortedInput input) {
 241         return null;
 242     }
 243
 244     @Override
 245     public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
 246             final RemoveShardReplicaInput input) {
 247         return null;
 248     }
 249
 250     @Override
 251     public ListenableFuture<RpcResult<SubscribeYnlOutput>> subscribeYnl(final SubscribeYnlInput input) {
 252         LOG.info("In subscribeYnl - input: {}", input);
 253
 254         if (ynlRegistrations.containsKey(input.getId())) {
 255             return RpcResultBuilder.<SubscribeYnlOutput>failed()
 256                 .withError(ErrorType.RPC, ErrorTag.DATA_EXISTS,
 257                     "There is already a listener registered for id: " + input.getId())
 258                 .buildFuture();
 259         }
 260
 261         ynlRegistrations.put(input.getId(),
 262                 notificationService.registerNotificationListener(new YnlListener(input.getId())));
 263
 264         return RpcResultBuilder.success(new SubscribeYnlOutputBuilder().build()).buildFuture();
 265     }
 266
 267
 268     @Override
 269     public ListenableFuture<RpcResult<UnregisterBoundConstantOutput>> unregisterBoundConstant(
 270             final UnregisterBoundConstantInput input) {
 271         LOG.info("In unregisterBoundConstant - {}", input);
 272
 273         final DOMRpcImplementationRegistration<RoutedGetConstantService> rpcRegistration =
 274                 routedRegistrations.remove(input.getContext());
 275
 276         if (rpcRegistration == null) {
 277             return RpcResultBuilder.<UnregisterBoundConstantOutput>failed()
 278                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING,
 279                     "No prior RPC was registered for " + input.getContext())
 280                 .buildFuture();
 281         }
 282
 283         rpcRegistration.close();
 284         return RpcResultBuilder.success(new UnregisterBoundConstantOutputBuilder().build()).buildFuture();
 285     }
 286
 287     @Override
 288     public ListenableFuture<RpcResult<RegisterSingletonConstantOutput>> registerSingletonConstant(
 289             final RegisterSingletonConstantInput input) {
 290         LOG.info("In registerSingletonConstant - input: {}", input);
 291
 292         if (input.getConstant() == null) {
 293             return RpcResultBuilder.<RegisterSingletonConstantOutput>failed()
 294                 .withError(ErrorType.RPC, ErrorTag.INVALID_VALUE, "Constant value is null")
 295                 .buildFuture();
 296         }
 297
 298         getSingletonConstantRegistration =
 299                 SingletonGetConstantService.registerNew(singletonService, domRpcService, input.getConstant());
 300
 301         return RpcResultBuilder.success(new RegisterSingletonConstantOutputBuilder().build()).buildFuture();
 302     }
 303
 304     @Override
 305     public ListenableFuture<RpcResult<RegisterDefaultConstantOutput>> registerDefaultConstant(
 306             final RegisterDefaultConstantInput input) {
 307         return null;
 308     }
 309
 310     @Override
 311     public ListenableFuture<RpcResult<UnregisterConstantOutput>> unregisterConstant(
 312             final UnregisterConstantInput input) {
 313         LOG.info("In unregisterConstant");
 314
 315         if (globalGetConstantRegistration == null) {
 316             return RpcResultBuilder.<UnregisterConstantOutput>failed()
 317                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING, "No prior RPC was registered")
 318                 .buildFuture();
 319         }
 320
 321         globalGetConstantRegistration.close();
 322         globalGetConstantRegistration = null;
 323
 324         return RpcResultBuilder.success(new UnregisterConstantOutputBuilder().build()).buildFuture();
 325     }
 326
 327     @Override
 328     public ListenableFuture<RpcResult<UnregisterFlappingSingletonOutput>> unregisterFlappingSingleton(
 329             final UnregisterFlappingSingletonInput input) {
 330         LOG.info("In unregisterFlappingSingleton");
 331
 332         if (flappingSingletonService == null) {
 333             return RpcResultBuilder.<UnregisterFlappingSingletonOutput>failed()
 334                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING, "No prior RPC was registered")
 335                 .buildFuture();
 336         }
 337
 338         final long flapCount = flappingSingletonService.setInactive();
 339         flappingSingletonService = null;
 340
 341         return RpcResultBuilder.success(new UnregisterFlappingSingletonOutputBuilder().setFlapCount(flapCount).build())
 342                 .buildFuture();
 343     }
 344
 345     @Override
 346     public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
 347         throw new UnsupportedOperationException();
 348     }
 349
 350     @Override
 351     public ListenableFuture<RpcResult<SubscribeDdtlOutput>> subscribeDdtl(final SubscribeDdtlInput input) {
 352         throw new UnsupportedOperationException();
 353     }
 354
 355     @Override
 356     public ListenableFuture<RpcResult<RegisterBoundConstantOutput>> registerBoundConstant(
 357             final RegisterBoundConstantInput input) {
 358         LOG.info("In registerBoundConstant - input: {}", input);
 359
 360         if (input.getContext() == null) {
 361             return RpcResultBuilder.<RegisterBoundConstantOutput>failed().withError(
 362                     ErrorType.RPC, ErrorTag.INVALID_VALUE, "Context value is null").buildFuture();
 363         }
 364
 365         if (input.getConstant() == null) {
 366             return RpcResultBuilder.<RegisterBoundConstantOutput>failed().withError(
 367                     ErrorType.RPC, ErrorTag.INVALID_VALUE, "Constant value is null").buildFuture();
 368         }
 369
 370         if (routedRegistrations.containsKey(input.getContext())) {
 371             return RpcResultBuilder.<RegisterBoundConstantOutput>failed().withError(ErrorType.RPC,
 372                 ErrorTag.DATA_EXISTS, "There is already an rpc registered for context: " + input.getContext())
 373                 .buildFuture();
 374         }
 375
 376         final DOMRpcImplementationRegistration<RoutedGetConstantService> rpcRegistration =
 377                 RoutedGetConstantService.registerNew(bindingNormalizedNodeSerializer, domRpcService,
 378                         input.getConstant(), input.getContext());
 379
 380         routedRegistrations.put(input.getContext(), rpcRegistration);
 381         return RpcResultBuilder.success(new RegisterBoundConstantOutputBuilder().build()).buildFuture();
 382     }
 383
 384     @Override
 385     public ListenableFuture<RpcResult<RegisterFlappingSingletonOutput>> registerFlappingSingleton(
 386             final RegisterFlappingSingletonInput input) {
 387         LOG.info("In registerFlappingSingleton");
 388
 389         if (flappingSingletonService != null) {
 390             return RpcResultBuilder.<RegisterFlappingSingletonOutput>failed()
 391                 .withError(ErrorType.RPC, ErrorTag.DATA_EXISTS, "There is already an rpc registered")
 392                 .buildFuture();
 393         }
 394
 395         flappingSingletonService = new FlappingSingletonService(singletonService);
 396
 397         return RpcResultBuilder.success(new RegisterFlappingSingletonOutputBuilder().build()).buildFuture();
 398     }
 399
 400     @Override
 401     public ListenableFuture<RpcResult<UnsubscribeDtclOutput>> unsubscribeDtcl(final UnsubscribeDtclInput input) {
 402         LOG.info("In unsubscribeDtcl");
 403
 404         if (idIntsListener == null || dtclReg == null) {
 405             return RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 406                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING, "No prior listener was registered")
 407                 .buildFuture();
 408         }
 409
 410         long timeout = 120L;
 411         try {
 412             idIntsListener.tryFinishProcessing().get(timeout, TimeUnit.SECONDS);
 413         } catch (InterruptedException | ExecutionException | TimeoutException e) {
 414             LOG.error("Unable to finish notification processing", e);
 415             return RpcResultBuilder.<UnsubscribeDtclOutput>failed().withError(ErrorType.APPLICATION,
 416                     "Unable to finish notification processing in " + timeout + " seconds", e).buildFuture();
 417         }
 418
 419         dtclReg.close();
 420         dtclReg = null;
 421
 422         if (!idIntsListener.hasTriggered()) {
 423             return RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 424                 .withError(ErrorType.APPLICATION, ErrorTag.OPERATION_FAILED,
 425                     "id-ints listener has not received any notifications.")
 426                 .buildFuture();
 427         }
 428
 429         try (DOMDataTreeReadTransaction rTx = domDataBroker.newReadOnlyTransaction()) {
 430             final Optional<NormalizedNode> readResult = rTx.read(LogicalDatastoreType.CONFIGURATION,
 431                 WriteTransactionsHandler.ID_INT_YID).get();
 432
 433             if (!readResult.isPresent()) {
 434                 return RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 435                     .withError(ErrorType.APPLICATION, ErrorTag.DATA_MISSING, "No data read from id-ints list")
 436                     .buildFuture();
 437             }
 438
 439             final boolean nodesEqual = idIntsListener.checkEqual(readResult.get());
 440             if (!nodesEqual) {
 441                 LOG.error("Final read of id-int does not match IdIntsListener's copy. {}",
 442                         idIntsListener.diffWithLocalCopy(readResult.get()));
 443             }
 444
 445             return RpcResultBuilder.success(new UnsubscribeDtclOutputBuilder().setCopyMatches(nodesEqual).build())
 446                     .buildFuture();
 447
 448         } catch (final InterruptedException | ExecutionException e) {
 449             LOG.error("Final read of id-ints failed", e);
 450             return RpcResultBuilder.<UnsubscribeDtclOutput>failed().withError(ErrorType.APPLICATION,
 451                     "Final read of id-ints failed", e).buildFuture();
 452         }
 453     }
 454
 455     @Override
 456     public ListenableFuture<RpcResult<UnsubscribeYnlOutput>> unsubscribeYnl(final UnsubscribeYnlInput input) {
 457         LOG.info("In unsubscribeYnl - input: {}", input);
 458
 459         if (!ynlRegistrations.containsKey(input.getId())) {
 460             return RpcResultBuilder.<UnsubscribeYnlOutput>failed()
 461                 .withError(ErrorType.RPC, ErrorTag.DATA_MISSING,
 462                     "No prior listener was registered for " + input.getId())
 463                 .buildFuture();
 464         }
 465
 466         final ListenerRegistration<YnlListener> reg = ynlRegistrations.remove(input.getId());
 467         final UnsubscribeYnlOutput output = reg.getInstance().getOutput();
 468
 469         reg.close();
 470
 471         return RpcResultBuilder.<UnsubscribeYnlOutput>success().withResult(output).buildFuture();
 472     }
 473
 474     @Override
 475     public ListenableFuture<RpcResult<CheckPublishNotificationsOutput>> checkPublishNotifications(
 476             final CheckPublishNotificationsInput input) {
 477         LOG.info("In checkPublishNotifications - input: {}", input);
 478
 479         final PublishNotificationsTask task = publishNotificationsTasks.get(input.getId());
 480
 481         if (task == null) {
 482             return RpcResultBuilder.success(new CheckPublishNotificationsOutputBuilder().setActive(false).build())
 483                 .buildFuture();
 484         }
 485
 486         final CheckPublishNotificationsOutputBuilder checkPublishNotificationsOutputBuilder =
 487                 new CheckPublishNotificationsOutputBuilder().setActive(!task.isFinished());
 488
 489         if (task.getLastError() != null) {
 490             LOG.error("Last error for {}", task, task.getLastError());
 491             checkPublishNotificationsOutputBuilder.setLastError(task.getLastError().toString());
 492         }
 493
 494         final CheckPublishNotificationsOutput output =
 495                 checkPublishNotificationsOutputBuilder.setPublishCount(task.getCurrentNotif()).build();
 496
 497         return RpcResultBuilder.success(output).buildFuture();
 498     }
 499
 500     @Override
 501     public ListenableFuture<RpcResult<ShutdownShardReplicaOutput>> shutdownShardReplica(
 502             final ShutdownShardReplicaInput input) {
 503         LOG.info("In shutdownShardReplica - input: {}", input);
 504
 505         final String shardName = input.getShardName();
 506         if (Strings.isNullOrEmpty(shardName)) {
 507             return RpcResultBuilder.<ShutdownShardReplicaOutput>failed()
 508                 .withError(ErrorType.RPC, ErrorTag.BAD_ELEMENT, shardName + "is not a valid shard name")
 509                 .buildFuture();
 510         }
 511
 512         return shutdownShardGracefully(shardName, new ShutdownShardReplicaOutputBuilder().build());
 513     }
 514
 515     private <T> SettableFuture<RpcResult<T>> shutdownShardGracefully(final String shardName, final T success) {
 516         final SettableFuture<RpcResult<T>> rpcResult = SettableFuture.create();
 517         final ActorUtils context = configDataStore.getActorUtils();
 518
 519         long timeoutInMS = Math.max(context.getDatastoreContext().getShardRaftConfig()
 520                 .getElectionTimeOutInterval().$times(3).toMillis(), 10000);
 521         final FiniteDuration duration = FiniteDuration.apply(timeoutInMS, TimeUnit.MILLISECONDS);
 522         final scala.concurrent.Promise<Boolean> shutdownShardAsk = Futures.promise();
 523
 524         context.findLocalShardAsync(shardName).onComplete(new OnComplete<ActorRef>() {
 525             @Override
 526             public void onComplete(final Throwable throwable, final ActorRef actorRef) {
 527                 if (throwable != null) {
 528                     shutdownShardAsk.failure(throwable);
 529                 } else {
 530                     shutdownShardAsk.completeWith(Patterns.gracefulStop(actorRef, duration, Shutdown.INSTANCE));
 531                 }
 532             }
 533         }, context.getClientDispatcher());
 534
 535         shutdownShardAsk.future().onComplete(new OnComplete<Boolean>() {
 536             @Override
 537             public void onComplete(final Throwable throwable, final Boolean gracefulStopResult) {
 538                 if (throwable != null) {
 539                     final RpcResult<T> failedResult = RpcResultBuilder.<T>failed()
 540                             .withError(ErrorType.APPLICATION, "Failed to gracefully shutdown shard", throwable).build();
 541                     rpcResult.set(failedResult);
 542                 } else {
 543                     // according to Patterns.gracefulStop API, we don't have to
 544                     // check value of gracefulStopResult
 545                     rpcResult.set(RpcResultBuilder.success(success).build());
 546                 }
 547             }
 548         }, context.getClientDispatcher());
 549         return rpcResult;
 550     }
 551
 552     @Override
 553     public ListenableFuture<RpcResult<RegisterConstantOutput>> registerConstant(final RegisterConstantInput input) {
 554         LOG.info("In registerConstant - input: {}", input);
 555
 556         if (input.getConstant() == null) {
 557             return RpcResultBuilder.<RegisterConstantOutput>failed()
 558                 .withError(ErrorType.RPC, ErrorTag.INVALID_VALUE, "Constant value is null")
 559                 .buildFuture();
 560         }
 561
 562         if (globalGetConstantRegistration != null) {
 563             return RpcResultBuilder.<RegisterConstantOutput>failed()
 564                 .withError(ErrorType.RPC, ErrorTag.DATA_EXISTS, "There is already an rpc registered")
 565                 .buildFuture();
 566         }
 567
 568         globalGetConstantRegistration = GetConstantService.registerNew(domRpcService, input.getConstant());
 569         return RpcResultBuilder.success(new RegisterConstantOutputBuilder().build()).buildFuture();
 570     }
 571
 572     @Override
 573     public ListenableFuture<RpcResult<UnregisterDefaultConstantOutput>> unregisterDefaultConstant(
 574             final UnregisterDefaultConstantInput input) {
 575         throw new UnsupportedOperationException();
 576     }
 577
 578     @Override
 579     @SuppressWarnings("checkstyle:IllegalCatch")
 580     public ListenableFuture<RpcResult<UnsubscribeDdtlOutput>> unsubscribeDdtl(final UnsubscribeDdtlInput input) {
 581         throw new UnsupportedOperationException();
 582     }
 583 }