restconf/sal-rest-connector/src/main/java/org/opendaylight/netconf/sal/restconf/impl/RestconfImpl.java

   1 /*
   2  * Copyright (c) 2014, 2015 Brocade Communication Systems, Inc., 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.netconf.sal.restconf.impl;
  10
  11 import com.google.common.base.CharMatcher;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.base.Splitter;
  15 import com.google.common.base.Strings;
  16 import com.google.common.collect.Iterables;
  17 import com.google.common.collect.Lists;
  18 import com.google.common.collect.Maps;
  19 import com.google.common.collect.Sets;
  20 import com.google.common.util.concurrent.CheckedFuture;
  21 import com.google.common.util.concurrent.FutureCallback;
  22 import com.google.common.util.concurrent.Futures;
  23 import java.net.URI;
  24 import java.net.URISyntaxException;
  25 import java.text.DateFormat;
  26 import java.text.ParseException;
  27 import java.text.SimpleDateFormat;
  28 import java.util.ArrayList;
  29 import java.util.Collection;
  30 import java.util.Collections;
  31 import java.util.Date;
  32 import java.util.HashMap;
  33 import java.util.HashSet;
  34 import java.util.Iterator;
  35 import java.util.List;
  36 import java.util.Map;
  37 import java.util.Map.Entry;
  38 import java.util.Set;
  39 import java.util.concurrent.CancellationException;
  40 import java.util.concurrent.CountDownLatch;
  41 import java.util.concurrent.ExecutionException;
  42 import javax.ws.rs.core.Context;
  43 import javax.ws.rs.core.Response;
  44 import javax.ws.rs.core.Response.ResponseBuilder;
  45 import javax.ws.rs.core.Response.Status;
  46 import javax.ws.rs.core.UriBuilder;
  47 import javax.ws.rs.core.UriInfo;
  48 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  49 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  50 import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
  51 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  52 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
  53 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
  54 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException;
  55 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  56 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
  57 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
  58 import org.opendaylight.netconf.md.sal.rest.common.RestconfValidationUtils;
  59 import org.opendaylight.netconf.sal.rest.api.Draft02;
  60 import org.opendaylight.netconf.sal.rest.api.RestconfService;
  61 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorTag;
  62 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorType;
  63 import org.opendaylight.netconf.sal.streams.listeners.ListenerAdapter;
  64 import org.opendaylight.netconf.sal.streams.listeners.NotificationListenerAdapter;
  65 import org.opendaylight.netconf.sal.streams.listeners.Notificator;
  66 import org.opendaylight.netconf.sal.streams.websockets.WebSocketServer;
  67 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  68 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  69 import org.opendaylight.yangtools.yang.common.QName;
  70 import org.opendaylight.yangtools.yang.common.QNameModule;
  71 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  73 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  74 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  75 import org.opendaylight.yangtools.yang.data.api.schema.AugmentationNode;
  76 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  77 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  78 import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
  79 import org.opendaylight.yangtools.yang.data.api.schema.LeafSetEntryNode;
  80 import org.opendaylight.yangtools.yang.data.api.schema.LeafSetNode;
  81 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  82 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  83 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  84 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  85 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  86 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
  87 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.DataContainerNodeAttrBuilder;
  88 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.ListNodeBuilder;
  89 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.NormalizedNodeAttrBuilder;
  90 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  91 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  92 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  93 import org.opendaylight.yangtools.yang.model.api.DataNodeContainer;
  94 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  95 import org.opendaylight.yangtools.yang.model.api.FeatureDefinition;
  96 import org.opendaylight.yangtools.yang.model.api.LeafListSchemaNode;
  97 import org.opendaylight.yangtools.yang.model.api.LeafSchemaNode;
  98 import org.opendaylight.yangtools.yang.model.api.ListSchemaNode;
  99 import org.opendaylight.yangtools.yang.model.api.Module;
 100 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
 101 import org.opendaylight.yangtools.yang.model.api.RpcDefinition;
 102 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 103 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
 104 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
 105 import org.opendaylight.yangtools.yang.parser.stmt.rfc6020.effective.EffectiveSchemaContext;
 106 import org.slf4j.Logger;
 107 import org.slf4j.LoggerFactory;
 108
 109 public class RestconfImpl implements RestconfService {
 110
 111     private static final RestconfImpl INSTANCE = new RestconfImpl();
 112
 113     /**
 114      * Notifications are served on port 8181.
 115      */
 116     private static final int NOTIFICATION_PORT = 8181;
 117
 118     private static final int CHAR_NOT_FOUND = -1;
 119
 120     private static final SimpleDateFormat REVISION_FORMAT = new SimpleDateFormat("yyyy-MM-dd");
 121
 122     private static final String SAL_REMOTE_NAMESPACE = "urn:opendaylight:params:xml:ns:yang:controller:md:sal:remote";
 123
 124     private BrokerFacade broker;
 125
 126     private ControllerContext controllerContext;
 127
 128     private static final Logger LOG = LoggerFactory.getLogger(RestconfImpl.class);
 129
 130     private static final DataChangeScope DEFAULT_SCOPE = DataChangeScope.BASE;
 131
 132     private static final LogicalDatastoreType DEFAULT_DATASTORE = LogicalDatastoreType.CONFIGURATION;
 133
 134     private static final URI NAMESPACE_EVENT_SUBSCRIPTION_AUGMENT = URI.create("urn:sal:restconf:event:subscription");
 135
 136     private static final String DATASTORE_PARAM_NAME = "datastore";
 137
 138     private static final String SCOPE_PARAM_NAME = "scope";
 139
 140     private static final String OUTPUT_TYPE_PARAM_NAME = "notification-output-type";
 141
 142     private static final String NETCONF_BASE = "urn:ietf:params:xml:ns:netconf:base:1.0";
 143
 144     private static final String NETCONF_BASE_PAYLOAD_NAME = "data";
 145
 146     private static final QName NETCONF_BASE_QNAME;
 147
 148     private static final QNameModule SAL_REMOTE_AUGMENT;
 149
 150     private static final YangInstanceIdentifier.AugmentationIdentifier SAL_REMOTE_AUG_IDENTIFIER;
 151
 152     public static final CharSequence DATA_SUBSCR = "data-change-event-subscription";
 153     private static final CharSequence CREATE_DATA_SUBSCR = "create-" + DATA_SUBSCR;
 154
 155     public static final CharSequence NOTIFICATION_STREAM = "notification-stream";
 156     private static final CharSequence CREATE_NOTIFICATION_STREAM = "create-" + NOTIFICATION_STREAM;
 157
 158     static {
 159         try {
 160             final Date eventSubscriptionAugRevision = new SimpleDateFormat("yyyy-MM-dd").parse("2014-07-08");
 161             NETCONF_BASE_QNAME = QName.create(QNameModule.create(new URI(NETCONF_BASE), null),
 162                     NETCONF_BASE_PAYLOAD_NAME);
 163             SAL_REMOTE_AUGMENT = QNameModule.create(NAMESPACE_EVENT_SUBSCRIPTION_AUGMENT,
 164                     eventSubscriptionAugRevision);
 165             SAL_REMOTE_AUG_IDENTIFIER = new YangInstanceIdentifier.AugmentationIdentifier(Sets.newHashSet(
 166                     QName.create(SAL_REMOTE_AUGMENT, "scope"),
 167                     QName.create(SAL_REMOTE_AUGMENT, "datastore"),
 168                     QName.create(SAL_REMOTE_AUGMENT, "notification-output-type")));
 169         } catch (final ParseException e) {
 170             final String errMsg = "It wasn't possible to convert revision date of sal-remote-augment to date";
 171             LOG.debug(errMsg);
 172             throw new RestconfDocumentedException(errMsg, ErrorType.APPLICATION, ErrorTag.OPERATION_FAILED);
 173         } catch (final URISyntaxException e) {
 174             final String errMsg = "It wasn't possible to create instance of URI class with "+NETCONF_BASE+" URI";
 175             throw new RestconfDocumentedException(errMsg, ErrorType.APPLICATION, ErrorTag.OPERATION_FAILED);
 176         }
 177     }
 178
 179     public void setBroker(final BrokerFacade broker) {
 180         this.broker = broker;
 181     }
 182
 183     public void setControllerContext(final ControllerContext controllerContext) {
 184         this.controllerContext = controllerContext;
 185     }
 186
 187     private RestconfImpl() {
 188     }
 189
 190     public static RestconfImpl getInstance() {
 191         return INSTANCE;
 192     }
 193
 194     @Override
 195     public NormalizedNodeContext getModules(final UriInfo uriInfo) {
 196         final Set<Module> allModules = this.controllerContext.getAllModules();
 197         final MapNode allModuleMap = makeModuleMapNode(allModules);
 198
 199         final SchemaContext schemaContext = this.controllerContext.getGlobalSchema();
 200
 201         final Module restconfModule = getRestconfModule();
 202         final DataSchemaNode modulesSchemaNode = this.controllerContext.getRestconfModuleRestConfSchemaNode(
 203                 restconfModule, Draft02.RestConfModule.MODULES_CONTAINER_SCHEMA_NODE);
 204         Preconditions.checkState(modulesSchemaNode instanceof ContainerSchemaNode);
 205
 206         final DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> moduleContainerBuilder =
 207                 Builders.containerBuilder((ContainerSchemaNode) modulesSchemaNode);
 208         moduleContainerBuilder.withChild(allModuleMap);
 209
 210         return new NormalizedNodeContext(new InstanceIdentifierContext<>(null, modulesSchemaNode,
 211                 null, schemaContext), moduleContainerBuilder.build(),
 212                 QueryParametersParser.parseWriterParameters(uriInfo));
 213     }
 214
 215     /**
 216      * Valid only for mount point
 217      */
 218     @Override
 219     public NormalizedNodeContext getModules(final String identifier, final UriInfo uriInfo) {
 220         Preconditions.checkNotNull(identifier);
 221         if ( ! identifier.contains(ControllerContext.MOUNT)) {
 222             final String errMsg = "URI has bad format. If modules behind mount point should be showed,"
 223                     + " URI has to end with " + ControllerContext.MOUNT;
 224             LOG.debug(errMsg + " for " + identifier);
 225             throw new RestconfDocumentedException(errMsg, ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 226         }
 227
 228         final InstanceIdentifierContext<?> mountPointIdentifier = this.controllerContext
 229                 .toMountPointIdentifier(identifier);
 230         final DOMMountPoint mountPoint = mountPointIdentifier.getMountPoint();
 231         final Set<Module> modules = this.controllerContext.getAllModules(mountPoint);
 232         final MapNode mountPointModulesMap = makeModuleMapNode(modules);
 233
 234         final Module restconfModule = getRestconfModule();
 235         final DataSchemaNode modulesSchemaNode = this.controllerContext.getRestconfModuleRestConfSchemaNode(
 236                 restconfModule, Draft02.RestConfModule.MODULES_CONTAINER_SCHEMA_NODE);
 237         Preconditions.checkState(modulesSchemaNode instanceof ContainerSchemaNode);
 238
 239         final DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> moduleContainerBuilder =
 240                 Builders.containerBuilder((ContainerSchemaNode) modulesSchemaNode);
 241         moduleContainerBuilder.withChild(mountPointModulesMap);
 242
 243         return new NormalizedNodeContext(new InstanceIdentifierContext<>(null, modulesSchemaNode,
 244                 mountPoint, this.controllerContext.getGlobalSchema()), moduleContainerBuilder.build(),
 245                 QueryParametersParser.parseWriterParameters(uriInfo));
 246     }
 247
 248     @Override
 249     public NormalizedNodeContext getModule(final String identifier, final UriInfo uriInfo) {
 250         Preconditions.checkNotNull(identifier);
 251         final QName moduleNameAndRevision = getModuleNameAndRevision(identifier);
 252         Module module = null;
 253         DOMMountPoint mountPoint = null;
 254         final SchemaContext schemaContext;
 255         if (identifier.contains(ControllerContext.MOUNT)) {
 256             final InstanceIdentifierContext<?> mountPointIdentifier = this.controllerContext
 257                     .toMountPointIdentifier(identifier);
 258             mountPoint = mountPointIdentifier.getMountPoint();
 259             module = this.controllerContext.findModuleByNameAndRevision(mountPoint, moduleNameAndRevision);
 260             schemaContext = mountPoint.getSchemaContext();
 261         } else {
 262             module = this.controllerContext.findModuleByNameAndRevision(moduleNameAndRevision);
 263             schemaContext = this.controllerContext.getGlobalSchema();
 264         }
 265
 266         if (module == null) {
 267             final String errMsg = "Module with name '" + moduleNameAndRevision.getLocalName()
 268                     + "' and revision '" + moduleNameAndRevision.getRevision() + "' was not found.";
 269             LOG.debug(errMsg);
 270             throw new RestconfDocumentedException(errMsg, ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT);
 271         }
 272
 273         final Module restconfModule = getRestconfModule();
 274         final Set<Module> modules = Collections.singleton(module);
 275         final MapNode moduleMap = makeModuleMapNode(modules);
 276
 277         final DataSchemaNode moduleSchemaNode = this.controllerContext.getRestconfModuleRestConfSchemaNode(
 278                 restconfModule, Draft02.RestConfModule.MODULE_LIST_SCHEMA_NODE);
 279         Preconditions.checkState(moduleSchemaNode instanceof ListSchemaNode);
 280
 281         return new NormalizedNodeContext(new InstanceIdentifierContext<>(null, moduleSchemaNode, mountPoint,
 282                 schemaContext), moduleMap, QueryParametersParser.parseWriterParameters(uriInfo));
 283     }
 284
 285     @Override
 286     public NormalizedNodeContext getAvailableStreams(final UriInfo uriInfo) {
 287         final SchemaContext schemaContext = this.controllerContext.getGlobalSchema();
 288         final Set<String> availableStreams = Notificator.getStreamNames();
 289         final Module restconfModule = getRestconfModule();
 290         final DataSchemaNode streamSchemaNode = this.controllerContext
 291                 .getRestconfModuleRestConfSchemaNode(restconfModule, Draft02.RestConfModule.STREAM_LIST_SCHEMA_NODE);
 292         Preconditions.checkState(streamSchemaNode instanceof ListSchemaNode);
 293
 294         final CollectionNodeBuilder<MapEntryNode, MapNode> listStreamsBuilder = Builders
 295                 .mapBuilder((ListSchemaNode) streamSchemaNode);
 296
 297         for (final String streamName : availableStreams) {
 298             listStreamsBuilder.withChild(toStreamEntryNode(streamName, streamSchemaNode));
 299         }
 300
 301         final DataSchemaNode streamsContainerSchemaNode = this.controllerContext.getRestconfModuleRestConfSchemaNode(
 302                 restconfModule, Draft02.RestConfModule.STREAMS_CONTAINER_SCHEMA_NODE);
 303         Preconditions.checkState(streamsContainerSchemaNode instanceof ContainerSchemaNode);
 304
 305         final DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> streamsContainerBuilder =
 306                 Builders.containerBuilder((ContainerSchemaNode) streamsContainerSchemaNode);
 307         streamsContainerBuilder.withChild(listStreamsBuilder.build());
 308
 309
 310         return new NormalizedNodeContext(new InstanceIdentifierContext<>(null, streamsContainerSchemaNode, null,
 311                 schemaContext), streamsContainerBuilder.build(), QueryParametersParser.parseWriterParameters(uriInfo));
 312     }
 313
 314     @Override
 315     public NormalizedNodeContext getOperations(final UriInfo uriInfo) {
 316         final Set<Module> allModules = this.controllerContext.getAllModules();
 317         return operationsFromModulesToNormalizedContext(allModules, null);
 318     }
 319
 320     @Override
 321     public NormalizedNodeContext getOperations(final String identifier, final UriInfo uriInfo) {
 322         Set<Module> modules = null;
 323         DOMMountPoint mountPoint = null;
 324         if (identifier.contains(ControllerContext.MOUNT)) {
 325             final InstanceIdentifierContext<?> mountPointIdentifier = this.controllerContext
 326                     .toMountPointIdentifier(identifier);
 327             mountPoint = mountPointIdentifier.getMountPoint();
 328             modules = this.controllerContext.getAllModules(mountPoint);
 329
 330         } else {
 331             final String errMsg = "URI has bad format. If operations behind mount point should be showed, URI has to "
 332                     + "end with ";
 333             LOG.debug(errMsg + ControllerContext.MOUNT + " for " + identifier);
 334             throw new RestconfDocumentedException(errMsg + ControllerContext.MOUNT, ErrorType.PROTOCOL,
 335                     ErrorTag.INVALID_VALUE);
 336         }
 337
 338         return operationsFromModulesToNormalizedContext(modules, mountPoint);
 339     }
 340
 341     /**
 342      * Special case only for GET restconf/operations use (since moment of pre-Beryllium
 343      * Yang parser and Yang model API removal). The method is creating fake
 344      * schema context with fake module and fake data by use own implementations
 345      * of schema nodes and module.
 346      *
 347      * @param modules
 348      *            - set of modules for get RPCs from every module
 349      * @param mountPoint
 350      *            - mount point, if in use otherwise null
 351      * @return {@link NormalizedNodeContext}
 352      */
 353     private NormalizedNodeContext operationsFromModulesToNormalizedContext(final Set<Module> modules,
 354             final DOMMountPoint mountPoint) {
 355
 356         final ContainerSchemaNodeImpl fakeCont = new ContainerSchemaNodeImpl();
 357         final List<LeafNode<Object>> listRpcNodes = new ArrayList<>();
 358         for (final Module m : modules) {
 359             for (final RpcDefinition rpc : m.getRpcs()) {
 360
 361                 final LeafSchemaNode fakeLeaf = new LeafSchemaNodeImpl(fakeCont.getPath(),
 362                         QName.create(ModuleImpl.moduleQName, m.getName() + ":" + rpc.getQName().getLocalName()));
 363                 fakeCont.addNodeChild(fakeLeaf);
 364                 listRpcNodes.add(Builders.leafBuilder(fakeLeaf).build());
 365             }
 366         }
 367         final ContainerSchemaNode fakeContSchNode = fakeCont;
 368         final DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> containerBuilder = Builders
 369                 .containerBuilder(fakeContSchNode);
 370
 371         for (final LeafNode<Object> rpcNode : listRpcNodes) {
 372             containerBuilder.withChild(rpcNode);
 373         }
 374
 375         final Module fakeModule = new ModuleImpl(fakeContSchNode);
 376
 377         final Set<Module> fakeModules = new HashSet<>();
 378         fakeModules.add(fakeModule);
 379         final SchemaContext fakeSchemaCtx = EffectiveSchemaContext.resolveSchemaContext(fakeModules);
 380         final InstanceIdentifierContext<ContainerSchemaNode> instanceIdentifierContext =
 381                 new InstanceIdentifierContext<>(null, fakeContSchNode, mountPoint, fakeSchemaCtx);
 382         return new NormalizedNodeContext(instanceIdentifierContext, containerBuilder.build());
 383     }
 384
 385     private Module getRestconfModule() {
 386         final Module restconfModule = this.controllerContext.getRestconfModule();
 387         if (restconfModule == null) {
 388             LOG.debug("ietf-restconf module was not found.");
 389             throw new RestconfDocumentedException("ietf-restconf module was not found.", ErrorType.APPLICATION,
 390                     ErrorTag.OPERATION_NOT_SUPPORTED);
 391         }
 392
 393         return restconfModule;
 394     }
 395
 396     private static QName getModuleNameAndRevision(final String identifier) {
 397         final int mountIndex = identifier.indexOf(ControllerContext.MOUNT);
 398         String moduleNameAndRevision = "";
 399         if (mountIndex >= 0) {
 400             moduleNameAndRevision = identifier.substring(mountIndex + ControllerContext.MOUNT.length());
 401         } else {
 402             moduleNameAndRevision = identifier;
 403         }
 404
 405         final Splitter splitter = Splitter.on("/").omitEmptyStrings();
 406         final Iterable<String> split = splitter.split(moduleNameAndRevision);
 407         final List<String> pathArgs = Lists.<String> newArrayList(split);
 408         if (pathArgs.size() < 2) {
 409             LOG.debug("URI has bad format. It should be \'moduleName/yyyy-MM-dd\' " + identifier);
 410             throw new RestconfDocumentedException(
 411                     "URI has bad format. End of URI should be in format \'moduleName/yyyy-MM-dd\'", ErrorType.PROTOCOL,
 412                     ErrorTag.INVALID_VALUE);
 413         }
 414
 415         try {
 416             final String moduleName = pathArgs.get(0);
 417             final String revision = pathArgs.get(1);
 418             final Date moduleRevision = REVISION_FORMAT.parse(revision);
 419             return QName.create(null, moduleRevision, moduleName);
 420         } catch (final ParseException e) {
 421             LOG.debug("URI has bad format. It should be \'moduleName/yyyy-MM-dd\' " + identifier);
 422             throw new RestconfDocumentedException("URI has bad format. It should be \'moduleName/yyyy-MM-dd\'",
 423                     ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 424         }
 425     }
 426
 427     @Override
 428     public Object getRoot() {
 429         return null;
 430     }
 431
 432     @Override
 433     public NormalizedNodeContext invokeRpc(final String identifier, final NormalizedNodeContext payload,
 434             final UriInfo uriInfo) {
 435         final SchemaPath type = payload.getInstanceIdentifierContext().getSchemaNode().getPath();
 436         final URI namespace = payload.getInstanceIdentifierContext().getSchemaNode().getQName().getNamespace();
 437         final CheckedFuture<DOMRpcResult, DOMRpcException> response;
 438         final DOMMountPoint mountPoint = payload.getInstanceIdentifierContext().getMountPoint();
 439         final SchemaContext schemaContext;
 440         if (mountPoint != null) {
 441             final Optional<DOMRpcService> mountRpcServices = mountPoint.getService(DOMRpcService.class);
 442             if ( ! mountRpcServices.isPresent()) {
 443                 LOG.debug("Error: Rpc service is missing.");
 444                 throw new RestconfDocumentedException("Rpc service is missing.");
 445             }
 446             schemaContext = mountPoint.getSchemaContext();
 447             response = mountRpcServices.get().invokeRpc(type, payload.getData());
 448         } else {
 449             if (namespace.toString().equals(SAL_REMOTE_NAMESPACE)) {
 450                 if (identifier.contains(CREATE_DATA_SUBSCR)) {
 451                     response = invokeSalRemoteRpcSubscribeRPC(payload);
 452                 } else if (identifier.contains(CREATE_NOTIFICATION_STREAM)) {
 453                     response = invokeSalRemoteRpcNotifiStrRPC(payload);
 454                 } else {
 455                     final String msg = "Not supported operation";
 456                     LOG.warn(msg);
 457                     throw new RestconfDocumentedException(msg, ErrorType.RPC, ErrorTag.OPERATION_NOT_SUPPORTED);
 458                 }
 459             } else {
 460                 response = this.broker.invokeRpc(type, payload.getData());
 461             }
 462             schemaContext = this.controllerContext.getGlobalSchema();
 463         }
 464
 465         final DOMRpcResult result = checkRpcResponse(response);
 466
 467         RpcDefinition resultNodeSchema = null;
 468         final NormalizedNode<?, ?> resultData = result.getResult();
 469         if ((result != null) && (result.getResult() != null)) {
 470             resultNodeSchema = (RpcDefinition) payload.getInstanceIdentifierContext().getSchemaNode();
 471         }
 472
 473         return new NormalizedNodeContext(new InstanceIdentifierContext<RpcDefinition>(null,
 474                 resultNodeSchema, mountPoint, schemaContext), resultData,
 475                 QueryParametersParser.parseWriterParameters(uriInfo));
 476     }
 477
 478     private static DOMRpcResult checkRpcResponse(final CheckedFuture<DOMRpcResult, DOMRpcException> response) {
 479         if (response == null) {
 480             return null;
 481         }
 482         try {
 483             final DOMRpcResult retValue = response.get();
 484             if ((retValue.getErrors() == null) || retValue.getErrors().isEmpty()) {
 485                 return retValue;
 486             }
 487             LOG.debug("RpcError message", retValue.getErrors());
 488             throw new RestconfDocumentedException("RpcError message", null, retValue.getErrors());
 489         } catch (final InterruptedException e) {
 490             final String errMsg = "The operation was interrupted while executing and did not complete.";
 491             LOG.debug("Rpc Interrupt - " + errMsg, e);
 492             throw new RestconfDocumentedException(errMsg, ErrorType.RPC, ErrorTag.PARTIAL_OPERATION);
 493         } catch (final ExecutionException e) {
 494             LOG.debug("Execution RpcError: ", e);
 495             Throwable cause = e.getCause();
 496             if (cause != null) {
 497                 while (cause.getCause() != null) {
 498                     cause = cause.getCause();
 499                 }
 500
 501                 if (cause instanceof IllegalArgumentException) {
 502                     throw new RestconfDocumentedException(cause.getMessage(), ErrorType.PROTOCOL,
 503                             ErrorTag.INVALID_VALUE);
 504                 } else if (cause instanceof DOMRpcImplementationNotAvailableException) {
 505                     throw new RestconfDocumentedException(cause.getMessage(), ErrorType.APPLICATION,
 506                             ErrorTag.OPERATION_NOT_SUPPORTED);
 507                 }
 508                 throw new RestconfDocumentedException("The operation encountered an unexpected error while executing.",
 509                         cause);
 510             } else {
 511                 throw new RestconfDocumentedException("The operation encountered an unexpected error while executing.",
 512                         e);
 513             }
 514         } catch (final CancellationException e) {
 515             final String errMsg = "The operation was cancelled while executing.";
 516             LOG.debug("Cancel RpcExecution: " + errMsg, e);
 517             throw new RestconfDocumentedException(errMsg, ErrorType.RPC, ErrorTag.PARTIAL_OPERATION);
 518         }
 519     }
 520
 521     private static void validateInput(final SchemaNode inputSchema, final NormalizedNodeContext payload) {
 522         if ((inputSchema != null) && (payload.getData() == null)) {
 523             // expected a non null payload
 524             throw new RestconfDocumentedException("Input is required.", ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
 525         } else if ((inputSchema == null) && (payload.getData() != null)) {
 526             // did not expect any input
 527             throw new RestconfDocumentedException("No input expected.", ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
 528         }
 529     }
 530
 531     private CheckedFuture<DOMRpcResult, DOMRpcException>
 532             invokeSalRemoteRpcSubscribeRPC(final NormalizedNodeContext payload) {
 533         final ContainerNode value = (ContainerNode) payload.getData();
 534         final QName rpcQName = payload.getInstanceIdentifierContext().getSchemaNode().getQName();
 535         final Optional<DataContainerChild<? extends PathArgument, ?>> path = value.getChild(new NodeIdentifier(
 536                 QName.create(payload.getInstanceIdentifierContext().getSchemaNode().getQName(), "path")));
 537         final Object pathValue = path.isPresent() ? path.get().getValue() : null;
 538
 539         if (!(pathValue instanceof YangInstanceIdentifier)) {
 540             final String errMsg = "Instance identifier was not normalized correctly ";
 541             LOG.debug(errMsg + rpcQName);
 542             throw new RestconfDocumentedException(errMsg, ErrorType.APPLICATION, ErrorTag.OPERATION_FAILED);
 543         }
 544
 545         final YangInstanceIdentifier pathIdentifier = ((YangInstanceIdentifier) pathValue);
 546         String streamName = (String) CREATE_DATA_SUBSCR;
 547         NotificationOutputType outputType = null;
 548         if (!pathIdentifier.isEmpty()) {
 549             final String fullRestconfIdentifier = DATA_SUBSCR
 550                     + this.controllerContext.toFullRestconfIdentifier(pathIdentifier, null);
 551
 552             LogicalDatastoreType datastore =
 553                     parseEnumTypeParameter(value, LogicalDatastoreType.class, DATASTORE_PARAM_NAME);
 554             datastore = datastore == null ? DEFAULT_DATASTORE : datastore;
 555
 556             DataChangeScope scope = parseEnumTypeParameter(value, DataChangeScope.class, SCOPE_PARAM_NAME);
 557             scope = scope == null ? DEFAULT_SCOPE : scope;
 558
 559             outputType = parseEnumTypeParameter(value, NotificationOutputType.class,
 560                     OUTPUT_TYPE_PARAM_NAME);
 561             outputType = outputType == null ? NotificationOutputType.XML : outputType;
 562
 563             streamName = Notificator.createStreamNameFromUri(fullRestconfIdentifier + "/datastore=" + datastore
 564                     + "/scope=" + scope);
 565         }
 566
 567         if (Strings.isNullOrEmpty(streamName)) {
 568             final String errMsg = "Path is empty or contains value node which is not Container or List build-in type.";
 569             LOG.debug(errMsg + pathIdentifier);
 570             throw new RestconfDocumentedException(errMsg, ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 571         }
 572
 573         final QName outputQname = QName.create(rpcQName, "output");
 574         final QName streamNameQname = QName.create(rpcQName, "stream-name");
 575
 576         final ContainerNode output =
 577                 ImmutableContainerNodeBuilder.create().withNodeIdentifier(new NodeIdentifier(outputQname))
 578                         .withChild(ImmutableNodes.leafNode(streamNameQname, streamName)).build();
 579
 580         if (!Notificator.existListenerFor(streamName)) {
 581             Notificator.createListener(pathIdentifier, streamName, outputType);
 582         }
 583
 584         final DOMRpcResult defaultDOMRpcResult = new DefaultDOMRpcResult(output);
 585
 586         return Futures.immediateCheckedFuture(defaultDOMRpcResult);
 587     }
 588
 589     @Override
 590     public NormalizedNodeContext invokeRpc(final String identifier, final String noPayload, final UriInfo uriInfo) {
 591         if ((noPayload != null) && !CharMatcher.WHITESPACE.matchesAllOf(noPayload)) {
 592             throw new RestconfDocumentedException("Content must be empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 593         }
 594
 595         String identifierEncoded = null;
 596         DOMMountPoint mountPoint = null;
 597         final SchemaContext schemaContext;
 598         if (identifier.contains(ControllerContext.MOUNT)) {
 599             // mounted RPC call - look up mount instance.
 600             final InstanceIdentifierContext<?> mountPointId = this.controllerContext.toMountPointIdentifier(identifier);
 601             mountPoint = mountPointId.getMountPoint();
 602             schemaContext = mountPoint.getSchemaContext();
 603             final int startOfRemoteRpcName = identifier.lastIndexOf(ControllerContext.MOUNT)
 604                     + ControllerContext.MOUNT.length() + 1;
 605             final String remoteRpcName = identifier.substring(startOfRemoteRpcName);
 606             identifierEncoded = remoteRpcName;
 607
 608         } else if (identifier.indexOf("/") != CHAR_NOT_FOUND) {
 609             final String slashErrorMsg = String.format("Identifier %n%s%ncan\'t contain slash "
 610                     + "character (/).%nIf slash is part of identifier name then use %%2F placeholder.", identifier);
 611             LOG.debug(slashErrorMsg);
 612             throw new RestconfDocumentedException(slashErrorMsg, ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 613         } else {
 614             identifierEncoded = identifier;
 615             schemaContext = this.controllerContext.getGlobalSchema();
 616         }
 617
 618         final String identifierDecoded = this.controllerContext.urlPathArgDecode(identifierEncoded);
 619
 620         RpcDefinition rpc = null;
 621         if (mountPoint == null) {
 622             rpc = this.controllerContext.getRpcDefinition(identifierDecoded, null);
 623         } else {
 624             rpc = findRpc(mountPoint.getSchemaContext(), identifierDecoded);
 625         }
 626
 627         if (rpc == null) {
 628             LOG.debug("RPC " + identifierDecoded + " does not exist.");
 629             throw new RestconfDocumentedException("RPC does not exist.", ErrorType.RPC, ErrorTag.UNKNOWN_ELEMENT);
 630         }
 631
 632         if (rpc.getInput() != null) {
 633             LOG.debug("RPC " + rpc + " does not need input value.");
 634             // FIXME : find a correct Error from specification
 635             throw new IllegalStateException("RPC " + rpc + " does'n need input value!");
 636         }
 637
 638         final CheckedFuture<DOMRpcResult, DOMRpcException> response;
 639         if (mountPoint != null) {
 640             final Optional<DOMRpcService> mountRpcServices = mountPoint.getService(DOMRpcService.class);
 641             if ( ! mountRpcServices.isPresent()) {
 642                 throw new RestconfDocumentedException("Rpc service is missing.");
 643             }
 644             response = mountRpcServices.get().invokeRpc(rpc.getPath(), null);
 645         } else {
 646             response = this.broker.invokeRpc(rpc.getPath(), null);
 647         }
 648
 649         final DOMRpcResult result = checkRpcResponse(response);
 650
 651         return new NormalizedNodeContext(new InstanceIdentifierContext<>(null, rpc, mountPoint, schemaContext),
 652                 result.getResult(), QueryParametersParser.parseWriterParameters(uriInfo));
 653     }
 654
 655     private static RpcDefinition findRpc(final SchemaContext schemaContext, final String identifierDecoded) {
 656         final String[] splittedIdentifier = identifierDecoded.split(":");
 657         if (splittedIdentifier.length != 2) {
 658             final String errMsg = identifierDecoded + " couldn't be splitted to 2 parts (module:rpc name)";
 659             LOG.debug(errMsg);
 660             throw new RestconfDocumentedException(errMsg, ErrorType.APPLICATION, ErrorTag.INVALID_VALUE);
 661         }
 662         for (final Module module : schemaContext.getModules()) {
 663             if (module.getName().equals(splittedIdentifier[0])) {
 664                 for (final RpcDefinition rpcDefinition : module.getRpcs()) {
 665                     if (rpcDefinition.getQName().getLocalName().equals(splittedIdentifier[1])) {
 666                         return rpcDefinition;
 667                     }
 668                 }
 669             }
 670         }
 671         return null;
 672     }
 673
 674     @Override
 675     public NormalizedNodeContext readConfigurationData(final String identifier, final UriInfo uriInfo) {
 676         final InstanceIdentifierContext<?> iiWithData = this.controllerContext.toInstanceIdentifier(identifier);
 677         final DOMMountPoint mountPoint = iiWithData.getMountPoint();
 678         NormalizedNode<?, ?> data = null;
 679         final YangInstanceIdentifier normalizedII = iiWithData.getInstanceIdentifier();
 680         if (mountPoint != null) {
 681             data = this.broker.readConfigurationData(mountPoint, normalizedII);
 682         } else {
 683             data = this.broker.readConfigurationData(normalizedII);
 684         }
 685         if(data == null) {
 686             final String errMsg = "Request could not be completed because the relevant data model content does not exist ";
 687             LOG.debug(errMsg + identifier);
 688             throw new RestconfDocumentedException(errMsg, ErrorType.APPLICATION, ErrorTag.DATA_MISSING);
 689         }
 690         return new NormalizedNodeContext(iiWithData, data, QueryParametersParser.parseWriterParameters(uriInfo));
 691     }
 692
 693     @Override
 694     public NormalizedNodeContext readOperationalData(final String identifier, final UriInfo uriInfo) {
 695         final InstanceIdentifierContext<?> iiWithData = this.controllerContext.toInstanceIdentifier(identifier);
 696         final DOMMountPoint mountPoint = iiWithData.getMountPoint();
 697         NormalizedNode<?, ?> data = null;
 698         final YangInstanceIdentifier normalizedII = iiWithData.getInstanceIdentifier();
 699         if (mountPoint != null) {
 700             data = this.broker.readOperationalData(mountPoint, normalizedII);
 701         } else {
 702             data = this.broker.readOperationalData(normalizedII);
 703         }
 704         if(data == null) {
 705             final String errMsg = "Request could not be completed because the relevant data model content does not exist ";
 706             LOG.debug(errMsg + identifier);
 707             throw new RestconfDocumentedException(errMsg , ErrorType.APPLICATION, ErrorTag.DATA_MISSING);
 708         }
 709         return new NormalizedNodeContext(iiWithData, data, QueryParametersParser.parseWriterParameters(uriInfo));
 710     }
 711
 712     @Override
 713     public Response updateConfigurationData(final String identifier, final NormalizedNodeContext payload) {
 714         Preconditions.checkNotNull(identifier);
 715         final InstanceIdentifierContext<?> iiWithData = payload.getInstanceIdentifierContext();
 716
 717         validateInput(iiWithData.getSchemaNode(), payload);
 718         validateTopLevelNodeName(payload, iiWithData.getInstanceIdentifier());
 719         validateListKeysEqualityInPayloadAndUri(payload);
 720
 721         final DOMMountPoint mountPoint = iiWithData.getMountPoint();
 722         final YangInstanceIdentifier normalizedII = iiWithData.getInstanceIdentifier();
 723
 724         /*
 725          * There is a small window where another write transaction could be updating the same data
 726          * simultaneously and we get an OptimisticLockFailedException. This error is likely
 727          * transient and The WriteTransaction#submit API docs state that a retry will likely
 728          * succeed. So we'll try again if that scenario occurs. If it fails a third time then it
 729          * probably will never succeed so we'll fail in that case.
 730          *
 731          * By retrying we're attempting to hide the internal implementation of the data store and
 732          * how it handles concurrent updates from the restconf client. The client has instructed us
 733          * to put the data and we should make every effort to do so without pushing optimistic lock
 734          * failures back to the client and forcing them to handle it via retry (and having to
 735          * document the behavior).
 736          */
 737         PutResult result = null;
 738         final TryOfPutData tryPutData = new TryOfPutData();
 739         while(true) {
 740             if (mountPoint != null) {
 741
 742                 result = this.broker.commitMountPointDataPut(mountPoint, normalizedII, payload.getData());
 743             } else {
 744                 result = this.broker.commitConfigurationDataPut(this.controllerContext.getGlobalSchema(), normalizedII,
 745                         payload.getData());
 746             }
 747             final CountDownLatch waiter = new CountDownLatch(1);
 748             Futures.addCallback(result.getFutureOfPutData(), new FutureCallback<Void>() {
 749
 750                 @Override
 751                 public void onSuccess(final Void result) {
 752                     handlingLoggerPut(null, tryPutData, identifier);
 753                     waiter.countDown();
 754                 }
 755
 756                 @Override
 757                 public void onFailure(final Throwable t) {
 758                     waiter.countDown();
 759                     handlingLoggerPut(t, tryPutData, identifier);
 760                 }
 761             });
 762
 763             try {
 764                 waiter.await();
 765             } catch (final Exception e) {
 766                 final String msg = "Problem while waiting for response";
 767                 LOG.warn(msg);
 768                 throw new RestconfDocumentedException(msg, e);
 769             }
 770
 771             if(tryPutData.isDone()){
 772                 break;
 773             } else {
 774                 throw new RestconfDocumentedException("Problem while PUT operations");
 775             }
 776         }
 777
 778         return Response.status(result.getStatus()).build();
 779     }
 780
 781     protected void handlingLoggerPut(final Throwable t, final TryOfPutData tryPutData, final String identifier) {
 782         if (t != null) {
 783             if (t instanceof OptimisticLockFailedException) {
 784                 if (tryPutData.countGet() <= 0) {
 785                     LOG.debug("Got OptimisticLockFailedException on last try - failing " + identifier);
 786                     throw new RestconfDocumentedException(t.getMessage(), t);
 787                 }
 788                 LOG.debug("Got OptimisticLockFailedException - trying again " + identifier);
 789                 tryPutData.countDown();
 790             } else {
 791                 LOG.debug("Update ConfigDataStore fail " + identifier, t);
 792                 throw new RestconfDocumentedException(t.getMessage(), t);
 793             }
 794         } else {
 795             LOG.trace("PUT Successful " + identifier);
 796             tryPutData.done();
 797         }
 798     }
 799
 800     private static void validateTopLevelNodeName(final NormalizedNodeContext node,
 801             final YangInstanceIdentifier identifier) {
 802
 803         final String payloadName = node.getData().getNodeType().getLocalName();
 804
 805         //no arguments
 806         if (identifier.isEmpty()) {
 807             //no "data" payload
 808             if (!node.getData().getNodeType().equals(NETCONF_BASE_QNAME)) {
 809                 throw new RestconfDocumentedException("Instance identifier has to contain at least one path argument",
 810                         ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
 811             }
 812         //any arguments
 813         } else {
 814             final String identifierName = identifier.getLastPathArgument().getNodeType().getLocalName();
 815             if (!payloadName.equals(identifierName)) {
 816                 throw new RestconfDocumentedException("Payload name (" + payloadName
 817                         + ") is different from identifier name (" + identifierName + ")", ErrorType.PROTOCOL,
 818                         ErrorTag.MALFORMED_MESSAGE);
 819             }
 820         }
 821     }
 822
 823     /**
 824      * Validates whether keys in {@code payload} are equal to values of keys in {@code iiWithData} for list schema node
 825      *
 826      * @throws RestconfDocumentedException
 827      *             if key values or key count in payload and URI isn't equal
 828      *
 829      */
 830     private static void validateListKeysEqualityInPayloadAndUri(final NormalizedNodeContext payload) {
 831         Preconditions.checkArgument(payload != null);
 832         final InstanceIdentifierContext<?> iiWithData = payload.getInstanceIdentifierContext();
 833         final PathArgument lastPathArgument = iiWithData.getInstanceIdentifier().getLastPathArgument();
 834         final SchemaNode schemaNode = iiWithData.getSchemaNode();
 835         final NormalizedNode<?, ?> data = payload.getData();
 836         if (schemaNode instanceof ListSchemaNode) {
 837             final List<QName> keyDefinitions = ((ListSchemaNode) schemaNode).getKeyDefinition();
 838             if ((lastPathArgument instanceof NodeIdentifierWithPredicates) && (data instanceof MapEntryNode)) {
 839                 final Map<QName, Object> uriKeyValues = ((NodeIdentifierWithPredicates) lastPathArgument).getKeyValues();
 840                 isEqualUriAndPayloadKeyValues(uriKeyValues, (MapEntryNode) data, keyDefinitions);
 841             }
 842         }
 843     }
 844
 845     private static void isEqualUriAndPayloadKeyValues(final Map<QName, Object> uriKeyValues,
 846             final MapEntryNode payload, final List<QName> keyDefinitions) {
 847
 848         final Map<QName, Object> mutableCopyUriKeyValues = Maps.newHashMap(uriKeyValues);
 849         for (final QName keyDefinition : keyDefinitions) {
 850             final Object uriKeyValue = mutableCopyUriKeyValues.remove(keyDefinition);
 851             // should be caught during parsing URI to InstanceIdentifier
 852             RestconfValidationUtils.checkDocumentedError(uriKeyValue != null, ErrorType.PROTOCOL, ErrorTag.DATA_MISSING,
 853                     "Missing key " + keyDefinition + " in URI.");
 854
 855             final Object dataKeyValue = payload.getIdentifier().getKeyValues().get(keyDefinition);
 856
 857             if ( ! uriKeyValue.equals(dataKeyValue)) {
 858                 final String errMsg = "The value '" + uriKeyValue + "' for key '" + keyDefinition.getLocalName() +
 859                         "' specified in the URI doesn't match the value '" + dataKeyValue
 860                         + "' specified in the message body. ";
 861                 throw new RestconfDocumentedException(errMsg, ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 862             }
 863         }
 864     }
 865
 866     @Override
 867     public Response createConfigurationData(final String identifier, final NormalizedNodeContext payload,
 868             final UriInfo uriInfo) {
 869        return createConfigurationData(payload, uriInfo);
 870     }
 871
 872     // FIXME create RestconfIdetifierHelper and move this method there
 873     private static YangInstanceIdentifier checkConsistencyOfNormalizedNodeContext(final NormalizedNodeContext payload) {
 874         Preconditions.checkArgument(payload != null);
 875         Preconditions.checkArgument(payload.getData() != null);
 876         Preconditions.checkArgument(payload.getData().getNodeType() != null);
 877         Preconditions.checkArgument(payload.getInstanceIdentifierContext() != null);
 878         Preconditions.checkArgument(payload.getInstanceIdentifierContext().getInstanceIdentifier() != null);
 879
 880         final QName payloadNodeQname = payload.getData().getNodeType();
 881         final YangInstanceIdentifier yangIdent = payload.getInstanceIdentifierContext().getInstanceIdentifier();
 882         if (payloadNodeQname.compareTo(yangIdent.getLastPathArgument().getNodeType()) > 0) {
 883             return yangIdent;
 884         }
 885         final InstanceIdentifierContext<?> parentContext = payload.getInstanceIdentifierContext();
 886         final SchemaNode parentSchemaNode = parentContext.getSchemaNode();
 887         if(parentSchemaNode instanceof DataNodeContainer) {
 888             final DataNodeContainer cast = (DataNodeContainer) parentSchemaNode;
 889             for (final DataSchemaNode child : cast.getChildNodes()) {
 890                 if (payloadNodeQname.compareTo(child.getQName()) == 0) {
 891                     return YangInstanceIdentifier.builder(yangIdent).node(child.getQName()).build();
 892                 }
 893             }
 894         }
 895         if (parentSchemaNode instanceof RpcDefinition) {
 896             return yangIdent;
 897         }
 898         final String errMsg = "Error parsing input: DataSchemaNode has not children ";
 899         LOG.info(errMsg + yangIdent);
 900         throw new RestconfDocumentedException(errMsg, ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
 901     }
 902
 903     @Override
 904     public Response createConfigurationData(final NormalizedNodeContext payload, final UriInfo uriInfo) {
 905         if (payload == null) {
 906             throw new RestconfDocumentedException("Input is required.", ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
 907         }
 908
 909         // FIXME: move this to parsing stage (we can have augmentation nodes here which do not have namespace)
 910 //        final URI payloadNS = payload.getData().getNodeType().getNamespace();
 911 //        if (payloadNS == null) {
 912 //            throw new RestconfDocumentedException(
 913 //                    "Data has bad format. Root element node must have namespace (XML format) or module name(JSON format)",
 914 //                    ErrorType.PROTOCOL, ErrorTag.UNKNOWN_NAMESPACE);
 915 //        }
 916
 917         final DOMMountPoint mountPoint = payload.getInstanceIdentifierContext().getMountPoint();
 918         final InstanceIdentifierContext<?> iiWithData = payload.getInstanceIdentifierContext();
 919         final YangInstanceIdentifier normalizedII = iiWithData.getInstanceIdentifier();
 920
 921         CheckedFuture<Void, TransactionCommitFailedException> future;
 922         if (mountPoint != null) {
 923             future = this.broker.commitConfigurationDataPost(mountPoint, normalizedII, payload.getData());
 924         } else {
 925             future = this.broker.commitConfigurationDataPost(this.controllerContext.getGlobalSchema(), normalizedII,
 926                     payload.getData());
 927         }
 928
 929         final CountDownLatch waiter = new CountDownLatch(1);
 930         Futures.addCallback(future, new FutureCallback<Void>() {
 931
 932             @Override
 933             public void onSuccess(final Void result) {
 934                 handlerLoggerPost(null, uriInfo);
 935                 waiter.countDown();
 936             }
 937
 938             @Override
 939             public void onFailure(final Throwable t) {
 940                 waiter.countDown();
 941                 handlerLoggerPost(t, uriInfo);
 942             }
 943         });
 944
 945         try {
 946             waiter.await();
 947         } catch (final Exception e) {
 948             final String msg = "Problem while waiting for response";
 949             LOG.warn(msg);
 950             throw new RestconfDocumentedException(msg, e);
 951         }
 952
 953         final ResponseBuilder responseBuilder = Response.status(Status.NO_CONTENT);
 954         // FIXME: Provide path to result.
 955         final URI location = resolveLocation(uriInfo, "", mountPoint, normalizedII);
 956         if (location != null) {
 957             responseBuilder.location(location);
 958         }
 959         return responseBuilder.build();
 960     }
 961
 962     protected void handlerLoggerPost(final Throwable t, final UriInfo uriInfo) {
 963         if (t != null) {
 964             final String errMsg = "Error creating data ";
 965             LOG.warn(errMsg + (uriInfo != null ? uriInfo.getPath() : ""), t);
 966             throw new RestconfDocumentedException(errMsg, t);
 967         } else {
 968             LOG.trace("Successfuly create data.");
 969         }
 970     }
 971
 972     private URI resolveLocation(final UriInfo uriInfo, final String uriBehindBase, final DOMMountPoint mountPoint,
 973             final YangInstanceIdentifier normalizedII) {
 974         if(uriInfo == null) {
 975             // This is null if invoked internally
 976             return null;
 977         }
 978
 979         final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 980         uriBuilder.path("config");
 981         try {
 982             uriBuilder.path(this.controllerContext.toFullRestconfIdentifier(normalizedII, mountPoint));
 983         } catch (final Exception e) {
 984             LOG.info("Location for instance identifier" + normalizedII + "wasn't created", e);
 985             return null;
 986         }
 987         return uriBuilder.build();
 988     }
 989
 990     @Override
 991     public Response deleteConfigurationData(final String identifier) {
 992         final InstanceIdentifierContext<?> iiWithData = this.controllerContext.toInstanceIdentifier(identifier);
 993         final DOMMountPoint mountPoint = iiWithData.getMountPoint();
 994         final YangInstanceIdentifier normalizedII = iiWithData.getInstanceIdentifier();
 995
 996         final CheckedFuture<Void, TransactionCommitFailedException> future;
 997         if (mountPoint != null) {
 998             future = this.broker.commitConfigurationDataDelete(mountPoint, normalizedII);
 999         } else {
1000             future = this.broker.commitConfigurationDataDelete(normalizedII);
1001         }
1002
1003         final CountDownLatch waiter = new CountDownLatch(1);
1004         Futures.addCallback(future, new FutureCallback<Void>() {
1005
1006             @Override
1007             public void onSuccess(final Void result) {
1008                 handlerLoggerDelete(null);
1009                 waiter.countDown();
1010             }
1011
1012             @Override
1013             public void onFailure(final Throwable t) {
1014                 waiter.countDown();
1015                 handlerLoggerDelete(t);
1016             }
1017
1018         });
1019
1020         try {
1021             waiter.await();
1022         } catch (final Exception e) {
1023             final String msg = "Problem while waiting for response";
1024             LOG.warn(msg);
1025             throw new RestconfDocumentedException(msg, e);
1026         }
1027
1028         return Response.status(Status.OK).build();
1029     }
1030
1031     protected void handlerLoggerDelete(final Throwable t) {
1032         if (t != null) {
1033             final String errMsg = "Error while deleting data";
1034             LOG.info(errMsg, t);
1035             throw new RestconfDocumentedException(errMsg, t);
1036         } else {
1037             LOG.trace("Successfuly delete data.");
1038         }
1039     }
1040
1041     /**
1042      * Subscribes to some path in schema context (stream) to listen on changes on this stream.
1043      *
1044      * Additional parameters for subscribing to stream are loaded via rpc input parameters:
1045      * <ul>
1046      * <li>datastore - default CONFIGURATION (other values of {@link LogicalDatastoreType} enum type)</li>
1047      * <li>scope - default BASE (other values of {@link DataChangeScope})</li>
1048      * </ul>
1049      */
1050     @Override
1051     public NormalizedNodeContext subscribeToStream(final String identifier, final UriInfo uriInfo) {
1052         boolean startTime_used = false;
1053         boolean stopTime_used = false;
1054         boolean filter_used = false;
1055         Date start = null;
1056         Date stop = null;
1057         String filter = null;
1058
1059         for (final Entry<String, List<String>> entry : uriInfo.getQueryParameters().entrySet()) {
1060             switch (entry.getKey()) {
1061                 case "start-time":
1062                     if (!startTime_used) {
1063                         startTime_used = true;
1064                         start = parseDateFromQueryParam(entry);
1065                     } else {
1066                         throw new RestconfDocumentedException("Start-time parameter can be used only once.");
1067                     }
1068                     break;
1069                 case "stop-time":
1070                     if (!stopTime_used) {
1071                         stopTime_used = true;
1072                         stop = parseDateFromQueryParam(entry);
1073                     } else {
1074                         throw new RestconfDocumentedException("Stop-time parameter can be used only once.");
1075                     }
1076                     break;
1077                 case "filter":
1078                     if (!filter_used) {
1079                         filter_used = true;
1080                         filter = entry.getValue().iterator().next();
1081                     } else {
1082                         throw new RestconfDocumentedException("Filter parameter can be used only once.");
1083                     }
1084                     break;
1085                 default:
1086                     throw new RestconfDocumentedException("Bad parameter used with notifications: " + entry.getKey());
1087             }
1088         }
1089         if(!startTime_used && stopTime_used){
1090             throw new RestconfDocumentedException("Stop-time parameter has to be used with start-time parameter.");
1091         }
1092         URI response = null;
1093         if (identifier.contains(DATA_SUBSCR)) {
1094             response = dataSubs(identifier, uriInfo, start, stop, filter);
1095         } else if (identifier.contains(NOTIFICATION_STREAM)) {
1096             response = notifStream(identifier, uriInfo, start, stop, filter);
1097         }
1098
1099         if(response != null){
1100             // prepare node with value of location
1101             final InstanceIdentifierContext<?> iid = prepareIIDSubsStreamOutput();
1102             final NormalizedNodeAttrBuilder<NodeIdentifier, Object, LeafNode<Object>> builder = ImmutableLeafNodeBuilder
1103                     .create().withValue(response.toString());
1104             builder.withNodeIdentifier(
1105                     NodeIdentifier.create(QName.create("subscribe:to:notification", "2016-10-28", "location")));
1106
1107             // prepare new header with location
1108             final Map<String, Object> headers = new HashMap<>();
1109             headers.put("Location", response);
1110
1111             return new NormalizedNodeContext(iid, builder.build(), headers);
1112         }
1113
1114         final String msg = "Bad type of notification of sal-remote";
1115         LOG.warn(msg);
1116         throw new RestconfDocumentedException(msg);
1117     }
1118
1119     private Date parseDateFromQueryParam(final Entry<String, List<String>> entry) {
1120         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
1121         String numOf_ms = "";
1122         final String value = event.getValue();
1123         if (value.contains(".")) {
1124             numOf_ms = numOf_ms + ".";
1125             final int lastChar = value.contains("Z") ? value.indexOf("Z") : (value.contains("+") ? value.indexOf("+")
1126                     : (value.contains("-") ? value.indexOf("-") : value.length()));
1127             for (int i = 0; i < (lastChar - value.indexOf(".") - 1); i++) {
1128                 numOf_ms = numOf_ms + "S";
1129             }
1130         }
1131         String zone = "";
1132         if (!value.contains("Z")) {
1133             zone = zone + "XXX";
1134         }
1135         final DateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss" + numOf_ms + zone);
1136
1137         try {
1138             return dateFormatter.parse(value.contains("Z") ? value.replace('T', ' ').substring(0, value.indexOf("Z"))
1139                     : value.replace('T', ' '));
1140         } catch (final ParseException e) {
1141             throw new RestconfDocumentedException("Cannot parse of value in date: " + value + e);
1142         }
1143     }
1144
1145     /**
1146      * @return {@link InstanceIdentifierContext} of location leaf for
1147      *         notification
1148      */
1149     private InstanceIdentifierContext<?> prepareIIDSubsStreamOutput() {
1150         final QName qnameBase = QName.create("subscribe:to:notification", "2016-10-28", "notifi");
1151         final SchemaContext schemaCtx = ControllerContext.getInstance().getGlobalSchema();
1152         final DataSchemaNode location = ((ContainerSchemaNode) schemaCtx
1153                 .findModuleByNamespaceAndRevision(qnameBase.getNamespace(), qnameBase.getRevision())
1154                 .getDataChildByName(qnameBase)).getDataChildByName(QName.create(qnameBase, "location"));
1155         final List<PathArgument> path = new ArrayList<>();
1156         path.add(NodeIdentifier.create(qnameBase));
1157         path.add(NodeIdentifier.create(QName.create(qnameBase, "location")));
1158
1159         return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location, null,
1160                 schemaCtx);
1161     }
1162
1163     /**
1164      * Register notification listener by stream name
1165      *
1166      * @param identifier
1167      *            - stream name
1168      * @param uriInfo
1169      *            - uriInfo
1170      * @param stop
1171      *            - stop-time of getting notification
1172      * @param start
1173      *            - start-time of getting notification
1174      * @param filter
1175      *            - indicate wh ich subset of allpossible events are of interest
1176      * @return {@link URI} of location
1177      */
1178     private URI notifStream(final String identifier, final UriInfo uriInfo, final Date start, final Date stop,
1179             final String filter) {
1180         final String streamName = Notificator.createStreamNameFromUri(identifier);
1181         if (Strings.isNullOrEmpty(streamName)) {
1182             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
1183         }
1184         final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
1185         if ((listeners == null) || listeners.isEmpty()) {
1186             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
1187                     ErrorTag.UNKNOWN_ELEMENT);
1188         }
1189
1190         for (final NotificationListenerAdapter listener : listeners) {
1191             this.broker.registerToListenNotification(listener);
1192             listener.setQueryParams(start, stop, filter);
1193         }
1194
1195         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
1196         int notificationPort = NOTIFICATION_PORT;
1197         try {
1198             final WebSocketServer webSocketServerInstance = WebSocketServer.getInstance();
1199             notificationPort = webSocketServerInstance.getPort();
1200         } catch (final NullPointerException e) {
1201             WebSocketServer.createInstance(NOTIFICATION_PORT);
1202         }
1203         final UriBuilder uriToWebsocketServerBuilder = uriBuilder.port(notificationPort).scheme("ws");
1204         final URI uriToWebsocketServer = uriToWebsocketServerBuilder.replacePath(streamName).build();
1205
1206         return uriToWebsocketServer;
1207     }
1208
1209     /**
1210      * Register data change listener by stream name
1211      *
1212      * @param identifier
1213      *            - stream name
1214      * @param uriInfo
1215      *            - uri info
1216      * @param stop
1217      *            - start-time of getting notification
1218      * @param start
1219      *            - stop-time of getting notification
1220      * @param filter
1221      *            - indicate which subset of all possible events are of interest
1222      * @return {@link URI} of location
1223      */
1224     private URI dataSubs(final String identifier, final UriInfo uriInfo, final Date start, final Date stop,
1225             final String filter) {
1226         final String streamName = Notificator.createStreamNameFromUri(identifier);
1227         if (Strings.isNullOrEmpty(streamName)) {
1228             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
1229         }
1230
1231         final ListenerAdapter listener = Notificator.getListenerFor(streamName);
1232         if (listener == null) {
1233             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT);
1234         }
1235         listener.setQueryParams(start, stop, filter);
1236
1237         final Map<String, String> paramToValues = resolveValuesFromUri(identifier);
1238         final LogicalDatastoreType datastore = parserURIEnumParameter(LogicalDatastoreType.class,
1239                 paramToValues.get(DATASTORE_PARAM_NAME));
1240         if (datastore == null) {
1241             throw new RestconfDocumentedException("Stream name doesn't contains datastore value (pattern /datastore=)",
1242                     ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
1243         }
1244         final DataChangeScope scope = parserURIEnumParameter(DataChangeScope.class, paramToValues.get(SCOPE_PARAM_NAME));
1245         if (scope == null) {
1246             throw new RestconfDocumentedException("Stream name doesn't contains datastore value (pattern /scope=)",
1247                     ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
1248         }
1249
1250         this.broker.registerToListenDataChanges(datastore, scope, listener);
1251
1252         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
1253         int notificationPort = NOTIFICATION_PORT;
1254         try {
1255             final WebSocketServer webSocketServerInstance = WebSocketServer.getInstance();
1256             notificationPort = webSocketServerInstance.getPort();
1257         } catch (final NullPointerException e) {
1258             WebSocketServer.createInstance(NOTIFICATION_PORT);
1259         }
1260         final UriBuilder uriToWebsocketServerBuilder = uriBuilder.port(notificationPort).scheme("ws");
1261         final URI uriToWebsocketServer = uriToWebsocketServerBuilder.replacePath(streamName).build();
1262
1263         return uriToWebsocketServer;
1264     }
1265
1266     @Override
1267     public PATCHStatusContext patchConfigurationData(final String identifier, final PATCHContext context,
1268             final UriInfo uriInfo) {
1269         if (context == null) {
1270             throw new RestconfDocumentedException("Input is required.", ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
1271         }
1272
1273         try {
1274             return this.broker.patchConfigurationDataWithinTransaction(context);
1275         } catch (final Exception e) {
1276             LOG.debug("Patch transaction failed", e);
1277             throw new RestconfDocumentedException(e.getMessage());
1278         }
1279     }
1280
1281     @Override
1282     public PATCHStatusContext patchConfigurationData(final PATCHContext context, @Context final UriInfo uriInfo) {
1283         if (context == null) {
1284             throw new RestconfDocumentedException("Input is required.", ErrorType.PROTOCOL, ErrorTag.MALFORMED_MESSAGE);
1285         }
1286
1287         try {
1288             return this.broker.patchConfigurationDataWithinTransaction(context);
1289         } catch (final Exception e) {
1290             LOG.debug("Patch transaction failed", e);
1291             throw new RestconfDocumentedException(e.getMessage());
1292         }
1293     }
1294
1295     /**
1296      * Load parameter for subscribing to stream from input composite node
1297      *
1298      * @param value
1299      *            contains value
1300      * @return enum object if its string value is equal to {@code paramName}. In other cases null.
1301      */
1302     private static <T> T parseEnumTypeParameter(final ContainerNode value, final Class<T> classDescriptor,
1303             final String paramName) {
1304         final Optional<DataContainerChild<? extends PathArgument, ?>> augNode = value.getChild(SAL_REMOTE_AUG_IDENTIFIER);
1305         if (!augNode.isPresent() && !(augNode instanceof AugmentationNode)) {
1306             return null;
1307         }
1308         final Optional<DataContainerChild<? extends PathArgument, ?>> enumNode =
1309                 ((AugmentationNode) augNode.get())
1310                         .getChild(new NodeIdentifier(QName.create(SAL_REMOTE_AUGMENT, paramName)));
1311         if (!enumNode.isPresent()) {
1312             return null;
1313         }
1314         final Object rawValue = enumNode.get().getValue();
1315         if (!(rawValue instanceof String)) {
1316             return null;
1317         }
1318
1319         return resolveAsEnum(classDescriptor, (String) rawValue);
1320     }
1321
1322     /**
1323      * Checks whether {@code value} is one of the string representation of enumeration {@code classDescriptor}
1324      *
1325      * @return enum object if string value of {@code classDescriptor} enumeration is equal to {@code value}. Other cases
1326      *         null.
1327      */
1328     private static <T> T parserURIEnumParameter(final Class<T> classDescriptor, final String value) {
1329         if (Strings.isNullOrEmpty(value)) {
1330             return null;
1331         }
1332         return resolveAsEnum(classDescriptor, value);
1333     }
1334
1335     private static <T> T resolveAsEnum(final Class<T> classDescriptor, final String value) {
1336         final T[] enumConstants = classDescriptor.getEnumConstants();
1337         if (enumConstants != null) {
1338             for (final T enm : classDescriptor.getEnumConstants()) {
1339                 if (((Enum<?>) enm).name().equals(value)) {
1340                     return enm;
1341                 }
1342             }
1343         }
1344         return null;
1345     }
1346
1347     private static Map<String, String> resolveValuesFromUri(final String uri) {
1348         final Map<String, String> result = new HashMap<>();
1349         final String[] tokens = uri.split("/");
1350         for (int i = 1; i < tokens.length; i++) {
1351             final String[] parameterTokens = tokens[i].split("=");
1352             if (parameterTokens.length == 2) {
1353                 result.put(parameterTokens[0], parameterTokens[1]);
1354             }
1355         }
1356         return result;
1357     }
1358
1359     private MapNode makeModuleMapNode(final Set<Module> modules) {
1360         Preconditions.checkNotNull(modules);
1361         final Module restconfModule = getRestconfModule();
1362         final DataSchemaNode moduleSchemaNode = this.controllerContext.getRestconfModuleRestConfSchemaNode(
1363                 restconfModule, Draft02.RestConfModule.MODULE_LIST_SCHEMA_NODE);
1364         Preconditions.checkState(moduleSchemaNode instanceof ListSchemaNode);
1365
1366         final CollectionNodeBuilder<MapEntryNode, MapNode> listModuleBuilder = Builders
1367                 .mapBuilder((ListSchemaNode) moduleSchemaNode);
1368
1369         for (final Module module : modules) {
1370             listModuleBuilder.withChild(toModuleEntryNode(module, moduleSchemaNode));
1371         }
1372         return listModuleBuilder.build();
1373     }
1374
1375     protected MapEntryNode toModuleEntryNode(final Module module, final DataSchemaNode moduleSchemaNode) {
1376         Preconditions.checkArgument(moduleSchemaNode instanceof ListSchemaNode,
1377                 "moduleSchemaNode has to be of type ListSchemaNode");
1378         final ListSchemaNode listModuleSchemaNode = (ListSchemaNode) moduleSchemaNode;
1379         final DataContainerNodeAttrBuilder<NodeIdentifierWithPredicates, MapEntryNode> moduleNodeValues = Builders
1380                 .mapEntryBuilder(listModuleSchemaNode);
1381
1382         List<DataSchemaNode> instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1383                 (listModuleSchemaNode), "name");
1384         final DataSchemaNode nameSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1385         Preconditions.checkState(nameSchemaNode instanceof LeafSchemaNode);
1386         moduleNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) nameSchemaNode).withValue(module.getName())
1387                 .build());
1388
1389         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1390                 (listModuleSchemaNode), "revision");
1391         final DataSchemaNode revisionSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1392         Preconditions.checkState(revisionSchemaNode instanceof LeafSchemaNode);
1393         final String revision = REVISION_FORMAT.format(module.getRevision());
1394         moduleNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) revisionSchemaNode).withValue(revision)
1395                 .build());
1396
1397         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1398                 (listModuleSchemaNode), "namespace");
1399         final DataSchemaNode namespaceSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1400         Preconditions.checkState(namespaceSchemaNode instanceof LeafSchemaNode);
1401         moduleNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) namespaceSchemaNode)
1402                 .withValue(module.getNamespace().toString()).build());
1403
1404         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1405                 (listModuleSchemaNode), "feature");
1406         final DataSchemaNode featureSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1407         Preconditions.checkState(featureSchemaNode instanceof LeafListSchemaNode);
1408         final ListNodeBuilder<Object, LeafSetEntryNode<Object>> featuresBuilder = Builders
1409                 .leafSetBuilder((LeafListSchemaNode) featureSchemaNode);
1410         for (final FeatureDefinition feature : module.getFeatures()) {
1411             featuresBuilder.withChild(Builders.leafSetEntryBuilder(((LeafListSchemaNode) featureSchemaNode))
1412                     .withValue(feature.getQName().getLocalName()).build());
1413         }
1414         moduleNodeValues.withChild(featuresBuilder.build());
1415
1416         return moduleNodeValues.build();
1417     }
1418
1419     protected MapEntryNode toStreamEntryNode(final String streamName, final DataSchemaNode streamSchemaNode) {
1420         Preconditions.checkArgument(streamSchemaNode instanceof ListSchemaNode,
1421                 "streamSchemaNode has to be of type ListSchemaNode");
1422         final ListSchemaNode listStreamSchemaNode = (ListSchemaNode) streamSchemaNode;
1423         final DataContainerNodeAttrBuilder<NodeIdentifierWithPredicates, MapEntryNode> streamNodeValues = Builders
1424                 .mapEntryBuilder(listStreamSchemaNode);
1425
1426         List<DataSchemaNode> instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1427                 (listStreamSchemaNode), "name");
1428         final DataSchemaNode nameSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1429         Preconditions.checkState(nameSchemaNode instanceof LeafSchemaNode);
1430         streamNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) nameSchemaNode).withValue(streamName)
1431                 .build());
1432
1433         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1434                 (listStreamSchemaNode), "description");
1435         final DataSchemaNode descriptionSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1436         Preconditions.checkState(descriptionSchemaNode instanceof LeafSchemaNode);
1437         streamNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) nameSchemaNode)
1438                 .withValue("DESCRIPTION_PLACEHOLDER").build());
1439
1440         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1441                 (listStreamSchemaNode), "replay-support");
1442         final DataSchemaNode replaySupportSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1443         Preconditions.checkState(replaySupportSchemaNode instanceof LeafSchemaNode);
1444         streamNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) replaySupportSchemaNode)
1445                 .withValue(Boolean.valueOf(true)).build());
1446
1447         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1448                 (listStreamSchemaNode), "replay-log-creation-time");
1449         final DataSchemaNode replayLogCreationTimeSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1450         Preconditions.checkState(replayLogCreationTimeSchemaNode instanceof LeafSchemaNode);
1451         streamNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) replayLogCreationTimeSchemaNode)
1452                 .withValue("").build());
1453
1454         instanceDataChildrenByName = ControllerContext.findInstanceDataChildrenByName(
1455                 (listStreamSchemaNode), "events");
1456         final DataSchemaNode eventsSchemaNode = Iterables.getFirst(instanceDataChildrenByName, null);
1457         Preconditions.checkState(eventsSchemaNode instanceof LeafSchemaNode);
1458         streamNodeValues.withChild(Builders.leafBuilder((LeafSchemaNode) eventsSchemaNode)
1459                 .withValue("").build());
1460
1461         return streamNodeValues.build();
1462     }
1463
1464     /**
1465      * Prepare stream for notification
1466      *
1467      * @param payload
1468      *            - contains list of qnames of notifications
1469      * @return - checked future object
1470      */
1471     private CheckedFuture<DOMRpcResult, DOMRpcException> invokeSalRemoteRpcNotifiStrRPC(
1472             final NormalizedNodeContext payload) {
1473         final ContainerNode data = (ContainerNode) payload.getData();
1474         LeafSetNode leafSet = null;
1475         String outputType = "XML";
1476         for (final DataContainerChild<? extends PathArgument, ?> dataChild : data.getValue()) {
1477             if (dataChild instanceof LeafSetNode) {
1478                 leafSet = (LeafSetNode) dataChild;
1479             } else if (dataChild instanceof AugmentationNode) {
1480                 outputType = (String) (((AugmentationNode) dataChild).getValue()).iterator().next().getValue();
1481             }
1482         }
1483
1484         final Collection<LeafSetEntryNode> entryNodes = leafSet.getValue();
1485         final List<SchemaPath> paths = new ArrayList<>();
1486         String streamName = CREATE_NOTIFICATION_STREAM + "/";
1487
1488         final Iterator<LeafSetEntryNode> iterator = entryNodes.iterator();
1489         while (iterator.hasNext()) {
1490             final QName valueQName = QName.create((String) iterator.next().getValue());
1491             final Module module = ControllerContext.getInstance()
1492                     .findModuleByNamespace(valueQName.getModule().getNamespace());
1493             Preconditions.checkNotNull(module, "Module for namespace " + valueQName.getModule().getNamespace()
1494                     + " does not exist");
1495             NotificationDefinition notifiDef = null;
1496             for (final NotificationDefinition notification : module.getNotifications()) {
1497                 if (notification.getQName().equals(valueQName)) {
1498                     notifiDef = notification;
1499                     break;
1500                 }
1501             }
1502             final String moduleName = module.getName();
1503             Preconditions.checkNotNull(notifiDef,
1504                     "Notification " + valueQName + "doesn't exist in module " + moduleName);
1505             paths.add(notifiDef.getPath());
1506             streamName = streamName + moduleName + ":" + valueQName.getLocalName();
1507             if (iterator.hasNext()) {
1508                 streamName = streamName + ",";
1509             }
1510         }
1511
1512         final QName rpcQName = payload.getInstanceIdentifierContext().getSchemaNode().getQName();
1513         final QName outputQname = QName.create(rpcQName, "output");
1514         final QName streamNameQname = QName.create(rpcQName, "notification-stream-identifier");
1515
1516         final ContainerNode output = ImmutableContainerNodeBuilder.create()
1517                 .withNodeIdentifier(new NodeIdentifier(outputQname))
1518                 .withChild(ImmutableNodes.leafNode(streamNameQname, streamName)).build();
1519
1520         if (!Notificator.existNotificationListenerFor(streamName)) {
1521             Notificator.createNotificationListener(paths, streamName, outputType);
1522         }
1523
1524         final DOMRpcResult defaultDOMRpcResult = new DefaultDOMRpcResult(output);
1525
1526         return Futures.immediateCheckedFuture(defaultDOMRpcResult);
1527     }
1528
1529     private class TryOfPutData {
1530         int tries = 2;
1531         boolean done = false;
1532
1533         void countDown() {
1534             this.tries--;
1535         }
1536
1537         void done() {
1538             this.done = true;
1539         }
1540
1541         boolean isDone() {
1542             return this.done;
1543         }
1544         int countGet() {
1545             return this.tries;
1546         }
1547     }
1548 }