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
   9 package org.opendaylight.controller.clustering.it.provider;
  10
  11 import static akka.actor.ActorRef.noSender;
  12
  13 import akka.actor.ActorRef;
  14 import akka.actor.ActorSystem;
  15 import akka.actor.PoisonPill;
  16 import akka.actor.Props;
  17 import com.google.common.base.Optional;
  18 import com.google.common.util.concurrent.CheckedFuture;
  19 import com.google.common.util.concurrent.Futures;
  20 import com.google.common.util.concurrent.SettableFuture;
  21 import java.io.PrintWriter;
  22 import java.io.StringWriter;
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.Map;
  26 import java.util.concurrent.Future;
  27 import java.util.concurrent.TimeUnit;
  28 import org.opendaylight.controller.cluster.ActorSystemProvider;
  29 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientLocalHistory;
  30 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
  31 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
  32 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
  33 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  34 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  35 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  36 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory;
  37 import org.opendaylight.controller.clustering.it.provider.impl.FlappingSingletonService;
  38 import org.opendaylight.controller.clustering.it.provider.impl.GetConstantService;
  39 import org.opendaylight.controller.clustering.it.provider.impl.IdIntsDOMDataTreeLIstener;
  40 import org.opendaylight.controller.clustering.it.provider.impl.IdIntsListener;
  41 import org.opendaylight.controller.clustering.it.provider.impl.PrefixLeaderHandler;
  42 import org.opendaylight.controller.clustering.it.provider.impl.PrefixShardHandler;
  43 import org.opendaylight.controller.clustering.it.provider.impl.ProduceTransactionsHandler;
  44 import org.opendaylight.controller.clustering.it.provider.impl.PublishNotificationsTask;
  45 import org.opendaylight.controller.clustering.it.provider.impl.RoutedGetConstantService;
  46 import org.opendaylight.controller.clustering.it.provider.impl.SingletonGetConstantService;
  47 import org.opendaylight.controller.clustering.it.provider.impl.WriteTransactionsHandler;
  48 import org.opendaylight.controller.clustering.it.provider.impl.YnlListener;
  49 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  50 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  51 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  52 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  53 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
  54 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
  55 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
  56 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationRegistration;
  57 import org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService;
  58 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  59 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  60 import org.opendaylight.controller.sal.core.api.model.SchemaService;
  61 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  62 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  63 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  64 import org.opendaylight.mdsal.dom.api.DOMDataTreeLoopException;
  65 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
  66 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  67 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  68 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.AddShardReplicaInput;
  69 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.BecomePrefixLeaderInput;
  70 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsInput;
  71 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsOutput;
  72 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CheckPublishNotificationsOutputBuilder;
  73 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.CreatePrefixShardInput;
  74 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.IsClientAbortedOutput;
  75 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.OdlMdsalLowlevelControlService;
  76 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ProduceTransactionsInput;
  77 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ProduceTransactionsOutput;
  78 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterBoundConstantInput;
  79 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterConstantInput;
  80 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterDefaultConstantInput;
  81 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RegisterSingletonConstantInput;
  82 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RemovePrefixShardInput;
  83 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.RemoveShardReplicaInput;
  84 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.StartPublishNotificationsInput;
  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.UnregisterBoundConstantInput;
  87 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterFlappingSingletonOutput;
  88 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnregisterFlappingSingletonOutputBuilder;
  89 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDdtlOutput;
  90 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDdtlOutputBuilder;
  91 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDtclOutput;
  92 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeDtclOutputBuilder;
  93 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlInput;
  94 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlOutput;
  95 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsInput;
  96 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutput;
  97 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  98 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  99 import org.opendaylight.yangtools.yang.common.RpcError;
 100 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
 101 import org.opendaylight.yangtools.yang.common.RpcResult;
 102 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 103 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 104 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 105 import org.slf4j.Logger;
 106 import org.slf4j.LoggerFactory;
 107
 108 public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService {
 109
 110     private static final Logger LOG = LoggerFactory.getLogger(MdsalLowLevelTestProvider.class);
 111     private static final org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType CONTROLLER_CONFIG =
 112             org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION;
 113
 114     private final RpcProviderRegistry rpcRegistry;
 115     private final BindingAwareBroker.RpcRegistration<OdlMdsalLowlevelControlService> registration;
 116     private final DistributedShardFactory distributedShardFactory;
 117     private final DistributedDataStoreInterface configDataStore;
 118     private final DOMDataTreeService domDataTreeService;
 119     private final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer;
 120     private final DOMDataBroker domDataBroker;
 121     private final NotificationPublishService notificationPublishService;
 122     private final NotificationService notificationService;
 123     private final SchemaService schemaService;
 124     private final ClusterSingletonServiceProvider singletonService;
 125     private final DOMRpcProviderService domRpcService;
 126     private final PrefixLeaderHandler prefixLeaderHandler;
 127     private final PrefixShardHandler prefixShardHandler;
 128     private final DOMDataTreeChangeService domDataTreeChangeService;
 129     private final ActorSystem actorSystem;
 130
 131     private Map<InstanceIdentifier<?>, DOMRpcImplementationRegistration<RoutedGetConstantService>> routedRegistrations =
 132             new HashMap<>();
 133
 134     private Map<String, ListenerRegistration<YnlListener>> ynlRegistrations = new HashMap<>();
 135
 136     private DOMRpcImplementationRegistration<GetConstantService> globalGetConstantRegistration = null;
 137     private ClusterSingletonServiceRegistration getSingletonConstantRegistration;
 138     private FlappingSingletonService flappingSingletonService;
 139     private ListenerRegistration<DOMDataTreeChangeListener> dtclReg;
 140     private IdIntsListener idIntsListener;
 141     private Map<String, PublishNotificationsTask> publishNotificationsTasks = new HashMap<>();
 142     private ListenerRegistration<IdIntsDOMDataTreeLIstener> ddtlReg;
 143     private IdIntsDOMDataTreeLIstener idIntsDdtl;
 144
 145
 146
 147     public MdsalLowLevelTestProvider(final RpcProviderRegistry rpcRegistry,
 148                                      final DOMRpcProviderService domRpcService,
 149                                      final ClusterSingletonServiceProvider singletonService,
 150                                      final SchemaService schemaService,
 151                                      final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer,
 152                                      final NotificationPublishService notificationPublishService,
 153                                      final NotificationService notificationService,
 154                                      final DOMDataBroker domDataBroker,
 155                                      final DOMDataTreeService domDataTreeService,
 156                                      final DistributedShardFactory distributedShardFactory,
 157                                      final DistributedDataStoreInterface configDataStore,
 158                                      final ActorSystemProvider actorSystemProvider) {
 159         this.rpcRegistry = rpcRegistry;
 160         this.domRpcService = domRpcService;
 161         this.singletonService = singletonService;
 162         this.schemaService = schemaService;
 163         this.bindingNormalizedNodeSerializer = bindingNormalizedNodeSerializer;
 164         this.notificationPublishService = notificationPublishService;
 165         this.notificationService = notificationService;
 166         this.domDataBroker = domDataBroker;
 167         this.domDataTreeService = domDataTreeService;
 168         this.distributedShardFactory = distributedShardFactory;
 169         this.configDataStore = configDataStore;
 170         this.actorSystem = actorSystemProvider.getActorSystem();
 171
 172         this.prefixLeaderHandler = new PrefixLeaderHandler(domDataTreeService, bindingNormalizedNodeSerializer);
 173
 174         domDataTreeChangeService =
 175                 (DOMDataTreeChangeService) domDataBroker.getSupportedExtensions().get(DOMDataTreeChangeService.class);
 176
 177         registration = rpcRegistry.addRpcImplementation(OdlMdsalLowlevelControlService.class, this);
 178
 179         prefixShardHandler = new PrefixShardHandler(distributedShardFactory, domDataTreeService,
 180                 bindingNormalizedNodeSerializer);
 181     }
 182
 183     @Override
 184     public Future<RpcResult<Void>> unregisterSingletonConstant() {
 185         LOG.debug("unregister-singleton-constant");
 186
 187         if (getSingletonConstantRegistration == null) {
 188             LOG.debug("No get-singleton-constant registration present.");
 189             final RpcError rpcError = RpcResultBuilder
 190                     .newError(ErrorType.APPLICATION, "missing-registration", "No get-singleton-constant rpc registration present.");
 191             final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
 192             return Futures.immediateFuture(result);
 193         }
 194
 195         try {
 196             getSingletonConstantRegistration.close();
 197             getSingletonConstantRegistration = null;
 198
 199             return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 200         } catch (final Exception e) {
 201             LOG.debug("There was a problem closing the singleton constant service", e);
 202             final RpcError rpcError = RpcResultBuilder
 203                     .newError(ErrorType.APPLICATION, "error-closing", "There was a problem closing get-singleton-constant");
 204             final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
 205             return Futures.immediateFuture(result);
 206         }
 207     }
 208
 209     @Override
 210     public Future<RpcResult<Void>> startPublishNotifications(final StartPublishNotificationsInput input) {
 211         LOG.debug("publish-notifications, input: {}", input);
 212
 213         final PublishNotificationsTask task = new PublishNotificationsTask(notificationPublishService, input.getId(),
 214                 input.getSeconds(), input.getNotificationsPerSecond());
 215
 216         publishNotificationsTasks.put(input.getId(), task);
 217
 218         task.start();
 219
 220         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 221     }
 222
 223     @Override
 224     public Future<RpcResult<Void>> subscribeDtcl() {
 225
 226         if (dtclReg != null) {
 227             final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
 228                     "There is already dataTreeChangeListener registered on id-ints list.");
 229             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 230         }
 231
 232         idIntsListener = new IdIntsListener();
 233
 234         dtclReg = domDataTreeChangeService
 235                 .registerDataTreeChangeListener(
 236                         new org.opendaylight.controller.md.sal.dom.api.DOMDataTreeIdentifier(
 237                                 CONTROLLER_CONFIG, WriteTransactionsHandler.ID_INT_YID),
 238                         idIntsListener);
 239
 240         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 241     }
 242
 243     @Override
 244     public Future<RpcResult<WriteTransactionsOutput>> writeTransactions(final WriteTransactionsInput input) {
 245         LOG.debug("write-transactions, input: {}", input);
 246
 247         final WriteTransactionsHandler writeTransactionsHandler = new WriteTransactionsHandler(domDataBroker, input);
 248
 249         final SettableFuture<RpcResult<WriteTransactionsOutput>> settableFuture = SettableFuture.create();
 250         writeTransactionsHandler.start(settableFuture);
 251
 252         return settableFuture;
 253     }
 254
 255     @Override
 256     public Future<RpcResult<IsClientAbortedOutput>> isClientAborted() {
 257         return null;
 258     }
 259
 260     @Override
 261     public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
 262         return null;
 263     }
 264
 265     @Override
 266     public Future<RpcResult<Void>> subscribeYnl(final SubscribeYnlInput input) {
 267
 268         LOG.debug("subscribe-ynl, input: {}", input);
 269
 270         if (ynlRegistrations.containsKey(input.getId())) {
 271             final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
 272                     "There is already ynl listener registered for this id: " + input.getId());
 273             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 274         }
 275
 276         ynlRegistrations.put(input.getId(),
 277                 notificationService.registerNotificationListener(new YnlListener(input.getId())));
 278
 279         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 280     }
 281
 282     @Override
 283     public Future<RpcResult<Void>> removePrefixShard(final RemovePrefixShardInput input) {
 284         LOG.debug("remove-prefix-shard, input: {}", input);
 285
 286         return prefixShardHandler.onRemovePrefixShard(input);
 287     }
 288
 289     @Override
 290     public Future<RpcResult<Void>> becomePrefixLeader(final BecomePrefixLeaderInput input) {
 291         LOG.debug("become-prefix-leader, input: {}", input);
 292
 293         return prefixLeaderHandler.makeLeaderLocal(input);
 294     }
 295
 296     @Override
 297     public Future<RpcResult<Void>> unregisterBoundConstant(final UnregisterBoundConstantInput input) {
 298         LOG.debug("unregister-bound-constant, {}", input);
 299
 300         final DOMRpcImplementationRegistration<RoutedGetConstantService> registration =
 301                 routedRegistrations.remove(input.getContext());
 302
 303         if (registration == null) {
 304             LOG.debug("No get-contexted-constant registration for context: {}", input.getContext());
 305             final RpcError rpcError = RpcResultBuilder
 306                     .newError(ErrorType.APPLICATION, "missing-registration", "No get-constant rpc registration present.");
 307             final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
 308             return Futures.immediateFuture(result);
 309         }
 310
 311         registration.close();
 312         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 313     }
 314
 315     @Override
 316     public Future<RpcResult<Void>> registerSingletonConstant(final RegisterSingletonConstantInput input) {
 317
 318         LOG.debug("Received register-singleton-constant rpc, input: {}", input);
 319
 320         if (input.getConstant() == null) {
 321             final RpcError error = RpcResultBuilder.newError(
 322                     ErrorType.RPC, "Invalid input.", "Constant value is null");
 323             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 324         }
 325
 326         getSingletonConstantRegistration =
 327                 SingletonGetConstantService.registerNew(singletonService, domRpcService, input.getConstant());
 328
 329         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 330     }
 331
 332     @Override
 333     public Future<RpcResult<Void>> registerDefaultConstant(RegisterDefaultConstantInput input) {
 334         return null;
 335     }
 336
 337     @Override
 338     public Future<RpcResult<Void>> unregisterConstant() {
 339
 340         if (globalGetConstantRegistration == null) {
 341             final RpcError rpcError = RpcResultBuilder
 342                     .newError(ErrorType.APPLICATION, "missing-registration", "No get-constant rpc registration present.");
 343             final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
 344             return Futures.immediateFuture(result);
 345         }
 346
 347         globalGetConstantRegistration.close();
 348         globalGetConstantRegistration = null;
 349
 350         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 351     }
 352
 353     @Override
 354     public Future<RpcResult<UnregisterFlappingSingletonOutput>> unregisterFlappingSingleton() {
 355         LOG.debug("unregister-flapping-singleton received.");
 356
 357         if (flappingSingletonService == null) {
 358             final RpcError rpcError = RpcResultBuilder
 359                     .newError(ErrorType.APPLICATION, "missing-registration", "No flapping-singleton registration present.");
 360             final RpcResult<UnregisterFlappingSingletonOutput> result =
 361                     RpcResultBuilder.<UnregisterFlappingSingletonOutput>failed().withRpcError(rpcError).build();
 362             return Futures.immediateFuture(result);
 363         }
 364
 365         final long flapCount = flappingSingletonService.setInactive();
 366         flappingSingletonService = null;
 367
 368         final UnregisterFlappingSingletonOutput output =
 369                 new UnregisterFlappingSingletonOutputBuilder().setFlapCount(flapCount).build();
 370
 371         return Futures.immediateFuture(RpcResultBuilder.success(output).build());
 372     }
 373
 374     @Override
 375     public Future<RpcResult<Void>> addShardReplica(AddShardReplicaInput input) {
 376         return null;
 377     }
 378
 379     @Override
 380     public Future<RpcResult<Void>> subscribeDdtl() {
 381
 382         if (ddtlReg != null) {
 383             final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
 384                     "There is already dataTreeChangeListener registered on id-ints list.");
 385             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 386         }
 387
 388         idIntsDdtl = new IdIntsDOMDataTreeLIstener();
 389
 390         try {
 391             ddtlReg =
 392                     domDataTreeService.registerListener(idIntsDdtl,
 393                             Collections.singleton(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 394                                     ProduceTransactionsHandler.ID_INT_YID))
 395                             , true, Collections.emptyList());
 396         } catch (DOMDataTreeLoopException e) {
 397             LOG.error("Failed to register DOMDataTreeListener.", e);
 398
 399         }
 400
 401         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 402     }
 403
 404     @Override
 405     public Future<RpcResult<Void>> registerBoundConstant(final RegisterBoundConstantInput input) {
 406         LOG.debug("register-bound-constant: {}", input);
 407
 408         if (input.getContext() == null) {
 409             final RpcError error = RpcResultBuilder.newError(
 410                     ErrorType.RPC, "Invalid input.", "Context value is null");
 411             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 412         }
 413
 414         if (input.getConstant() == null) {
 415             final RpcError error = RpcResultBuilder.newError(
 416                     ErrorType.RPC, "Invalid input.", "Constant value is null");
 417             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 418         }
 419
 420         if (routedRegistrations.containsKey(input.getContext())) {
 421             final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
 422                     "There is already a rpc registered for context: " + input.getContext());
 423             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 424         }
 425
 426         final DOMRpcImplementationRegistration<RoutedGetConstantService> registration =
 427                 RoutedGetConstantService.registerNew(bindingNormalizedNodeSerializer, domRpcService,
 428                         input.getConstant(), input.getContext());
 429
 430         routedRegistrations.put(input.getContext(), registration);
 431         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 432     }
 433
 434     @Override
 435     public Future<RpcResult<Void>> registerFlappingSingleton() {
 436         LOG.debug("Received register-flapping-singleton.");
 437
 438         if (flappingSingletonService != null) {
 439             final RpcError error = RpcResultBuilder.newError(
 440                     ErrorType.RPC, "Registration present.", "flapping-singleton already registered");
 441             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 442         }
 443
 444         flappingSingletonService = new FlappingSingletonService(singletonService);
 445
 446         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 447     }
 448
 449     @Override
 450     public Future<RpcResult<UnsubscribeDtclOutput>> unsubscribeDtcl() {
 451         LOG.debug("Received unsubscribe-dtcl");
 452
 453         if (idIntsListener == null || dtclReg == null) {
 454             final RpcError error = RpcResultBuilder.newError(
 455                     ErrorType.RPC, "Dtcl missing.", "No DataTreeChangeListener registered.");
 456             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed().withRpcError(error).build());
 457         }
 458
 459         dtclReg.close();
 460         dtclReg = null;
 461
 462         if (!idIntsListener.hasTriggered()) {
 463             final RpcError error = RpcResultBuilder.newError(
 464                     ErrorType.APPLICATION, "No notification received.", "id-ints listener has not received" +
 465                             "any notifications.");
 466             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 467                     .withRpcError(error).build());
 468         }
 469
 470         final DOMDataReadOnlyTransaction rTx = domDataBroker.newReadOnlyTransaction();
 471         try {
 472             final Optional<NormalizedNode<?, ?>> readResult =
 473                     rTx.read(CONTROLLER_CONFIG, WriteTransactionsHandler.ID_INT_YID).checkedGet();
 474
 475             if (!readResult.isPresent()) {
 476                 final RpcError error = RpcResultBuilder.newError(
 477                         ErrorType.APPLICATION, "Final read empty.", "No data read from id-ints list.");
 478                 return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 479                         .withRpcError(error).build());
 480             }
 481
 482             return Futures.immediateFuture(
 483                     RpcResultBuilder.success(new UnsubscribeDtclOutputBuilder()
 484                             .setCopyMatches(idIntsListener.checkEqual(readResult.get()))).build());
 485
 486         } catch (final ReadFailedException e) {
 487             final RpcError error = RpcResultBuilder.newError(
 488                     ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed.");
 489             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
 490                     .withRpcError(error).build());
 491
 492         }
 493     }
 494
 495     @Override
 496     public Future<RpcResult<Void>> createPrefixShard(final CreatePrefixShardInput input) {
 497         LOG.debug("create-prefix-shard, input: {}", input);
 498
 499         return prefixShardHandler.onCreatePrefixShard(input);
 500     }
 501
 502     @Override
 503     public Future<RpcResult<Void>> deconfigureIdIntsShard() {
 504         return null;
 505     }
 506
 507     @Override
 508     public Future<RpcResult<UnsubscribeYnlOutput>> unsubscribeYnl(final UnsubscribeYnlInput input) {
 509         LOG.debug("Received unsubscribe-ynl, input: {}", input);
 510
 511         if (!ynlRegistrations.containsKey(input.getId())) {
 512             final RpcError rpcError = RpcResultBuilder
 513                     .newError(ErrorType.APPLICATION, "missing-registration", "No ynl listener with this id registered.");
 514             final RpcResult<UnsubscribeYnlOutput> result =
 515                     RpcResultBuilder.<UnsubscribeYnlOutput>failed().withRpcError(rpcError).build();
 516             return Futures.immediateFuture(result);
 517         }
 518
 519         final ListenerRegistration<YnlListener> registration = ynlRegistrations.remove(input.getId());
 520         final UnsubscribeYnlOutput output = registration.getInstance().getOutput();
 521
 522         registration.close();
 523
 524         return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeYnlOutput>success().withResult(output).build());
 525     }
 526
 527     @Override
 528     public Future<RpcResult<CheckPublishNotificationsOutput>> checkPublishNotifications(
 529             final CheckPublishNotificationsInput input) {
 530
 531         final PublishNotificationsTask task = publishNotificationsTasks.get(input.getId());
 532
 533         if (task == null) {
 534             return Futures.immediateFuture(RpcResultBuilder.success(
 535                     new CheckPublishNotificationsOutputBuilder().setActive(false)).build());
 536         }
 537
 538         final CheckPublishNotificationsOutputBuilder checkPublishNotificationsOutputBuilder =
 539                 new CheckPublishNotificationsOutputBuilder().setActive(!task.isFinished());
 540
 541         if (task.getLastError() != null) {
 542             final StringWriter sw = new StringWriter();
 543             final PrintWriter pw = new PrintWriter(sw);
 544             task.getLastError().printStackTrace(pw);
 545             checkPublishNotificationsOutputBuilder.setLastError(task.getLastError().toString() + sw.toString());
 546         }
 547
 548         final CheckPublishNotificationsOutput output =
 549                 checkPublishNotificationsOutputBuilder.setPublishCount(task.getCurrentNotif()).build();
 550
 551         return Futures.immediateFuture(RpcResultBuilder.success(output).build());
 552     }
 553
 554     @Override
 555     public Future<RpcResult<ProduceTransactionsOutput>> produceTransactions(final ProduceTransactionsInput input) {
 556         LOG.debug("producer-transactions, input: {}", input);
 557
 558         final ProduceTransactionsHandler handler =
 559                 new ProduceTransactionsHandler(domDataTreeService, input);
 560
 561         final SettableFuture<RpcResult<ProduceTransactionsOutput>> settableFuture = SettableFuture.create();
 562         handler.start(settableFuture);
 563
 564         return settableFuture;
 565     }
 566
 567     @Override
 568     public Future<RpcResult<Void>> registerConstant(final RegisterConstantInput input) {
 569
 570         LOG.debug("Received register-constant rpc, input: {}", input);
 571
 572         if (input.getConstant() == null) {
 573             final RpcError error = RpcResultBuilder.newError(
 574                     ErrorType.RPC, "Invalid input.", "Constant value is null");
 575             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 576         }
 577
 578         if (globalGetConstantRegistration != null) {
 579             final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
 580                     "There is already a get-constant rpc registered.");
 581             return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
 582         }
 583
 584         globalGetConstantRegistration = GetConstantService.registerNew(domRpcService, input.getConstant());
 585         return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 586     }
 587
 588     @Override
 589     public Future<RpcResult<Void>> unregisterDefaultConstant() {
 590         return null;
 591     }
 592
 593     @Override
 594     public Future<RpcResult<UnsubscribeDdtlOutput>> unsubscribeDdtl() {
 595         LOG.debug("Received unsubscribe-ddtl.");
 596
 597         if (idIntsDdtl == null || ddtlReg == null) {
 598             final RpcError error = RpcResultBuilder.newError(
 599                     ErrorType.RPC, "Ddtl missing.", "No DOMDataTreeListener registered.");
 600             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed().withRpcError(error).build());
 601         }
 602
 603         ddtlReg.close();
 604         ddtlReg = null;
 605
 606         if (!idIntsDdtl.hasTriggered()) {
 607             final RpcError error = RpcResultBuilder.newError(
 608                     ErrorType.APPLICATION, "No notification received.", "id-ints listener has not received" +
 609                             "any notifications.");
 610             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
 611                     .withRpcError(error).build());
 612         }
 613
 614         final String shardName = ClusterUtils.getCleanShardName(ProduceTransactionsHandler.ID_INTS_YID);
 615         LOG.debug("Creating distributed datastore client for shard {}", shardName);
 616
 617         final ActorContext actorContext = configDataStore.getActorContext();
 618         final Props distributedDataStoreClientProps =
 619                 SimpleDataStoreClientActor.props(actorContext.getCurrentMemberName(),
 620                         "Shard-" + shardName, actorContext, shardName);
 621
 622         final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
 623         final DataStoreClient distributedDataStoreClient;
 624         try {
 625             distributedDataStoreClient = SimpleDataStoreClientActor
 626                     .getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
 627         } catch (final Exception e) {
 628             LOG.error("Failed to get actor for {}", distributedDataStoreClientProps, e);
 629             clientActor.tell(PoisonPill.getInstance(), noSender());
 630             final RpcError error = RpcResultBuilder.newError(
 631                     ErrorType.APPLICATION, "Unable to create ds client for read.",
 632                     "Unable to create ds client for read.");
 633             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
 634                     .withRpcError(error).build());
 635         }
 636
 637         final ClientLocalHistory localHistory = distributedDataStoreClient.createLocalHistory();
 638         final ClientTransaction tx = localHistory.createTransaction();
 639         final CheckedFuture<Optional<NormalizedNode<?, ?>>,
 640                 org.opendaylight.mdsal.common.api.ReadFailedException> read =
 641                 tx.read(YangInstanceIdentifier.of(ProduceTransactionsHandler.ID_INT));
 642
 643         tx.abort();
 644         localHistory.close();
 645         try {
 646             final Optional<NormalizedNode<?, ?>> optional = read.checkedGet();
 647             if (!optional.isPresent()) {
 648                 LOG.warn("Final read from client is empty.");
 649                 final RpcError error = RpcResultBuilder.newError(
 650                         ErrorType.APPLICATION, "Read failed.", "Final read from id-ints is empty.");
 651                 return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
 652                         .withRpcError(error).build());
 653             }
 654
 655             return Futures.immediateFuture(
 656                     RpcResultBuilder.success(new UnsubscribeDdtlOutputBuilder()
 657                             .setCopyMatches(idIntsDdtl.checkEqual(optional.get()))).build());
 658
 659         } catch (org.opendaylight.mdsal.common.api.ReadFailedException e) {
 660             LOG.error("Unable to read data to verify ddtl data.", e);
 661             final RpcError error = RpcResultBuilder.newError(
 662                     ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed.");
 663             return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
 664                     .withRpcError(error).build());
 665         } finally {
 666             distributedDataStoreClient.close();
 667             clientActor.tell(PoisonPill.getInstance(), noSender());
 668         }
 669     }
 670 }