restconf/restconf-nb/src/main/java/org/opendaylight/restconf/nb/rfc8040/rests/services/impl/CreateStreamUtil.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.restconf.nb.rfc8040.rests.services.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.collect.ImmutableSet;
  13 import java.util.Collection;
  14 import java.util.Optional;
  15 import java.util.Set;
  16 import java.util.stream.Collectors;
  17 import org.eclipse.jdt.annotation.Nullable;
  18 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  19 import org.opendaylight.mdsal.dom.api.DOMMountPoint;
  20 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  21 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
  22 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  23 import org.opendaylight.mdsal.dom.api.DOMSchemaService;
  24 import org.opendaylight.mdsal.dom.spi.DefaultDOMRpcResult;
  25 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  26 import org.opendaylight.restconf.nb.rfc8040.legacy.NormalizedNodePayload;
  27 import org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants;
  28 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.DeviceNotificationListenerAdaptor;
  29 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
  30 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  31 import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.device.notification.rev221106.SubscribeDeviceNotificationInput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.device.notification.rev221106.SubscribeDeviceNotificationOutput;
  34 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.CreateDataChangeEventSubscriptionInput1.Scope;
  35 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping;
  36 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  37 import org.opendaylight.yangtools.yang.common.ErrorTag;
  38 import org.opendaylight.yangtools.yang.common.ErrorType;
  39 import org.opendaylight.yangtools.yang.common.QName;
  40 import org.opendaylight.yangtools.yang.common.QNameModule;
  41 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  42 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.AugmentationIdentifier;
  43 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  44 import org.opendaylight.yangtools.yang.data.api.schema.AugmentationNode;
  45 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  46 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  47 import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
  48 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  49 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  50 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  51 import org.opendaylight.yangtools.yang.model.api.Module;
  52 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
  53 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  54 import org.slf4j.Logger;
  55 import org.slf4j.LoggerFactory;
  56
  57 /**
  58  * Utility class for creation of data-change-event or YANG notification streams.
  59  */
  60 final class CreateStreamUtil {
  61     private static final Logger LOG = LoggerFactory.getLogger(CreateStreamUtil.class);
  62     private static final QNameModule SAL_REMOTE_AUGMENT = NotificationOutputTypeGrouping.QNAME.getModule();
  63
  64     private static final QNameModule DEVICE_NOTIFICATION_MODULE = SubscribeDeviceNotificationInput.QNAME.getModule();
  65     private static final QName DATASTORE_QNAME =
  66         QName.create(SAL_REMOTE_AUGMENT, RestconfStreamsConstants.DATASTORE_PARAM_NAME).intern();
  67     private static final QName SCOPE_QNAME =
  68         QName.create(SAL_REMOTE_AUGMENT, RestconfStreamsConstants.SCOPE_PARAM_NAME).intern();
  69     private static final QName OUTPUT_TYPE_QNAME =
  70         QName.create(SAL_REMOTE_AUGMENT, "notification-output-type").intern();
  71     private static final QName DEVICE_NOTIFICATION_PATH_QNAME =
  72         QName.create(DEVICE_NOTIFICATION_MODULE, "path").intern();
  73     private static final QName DEVICE_NOTIFICATION_STREAM_PATH =
  74         QName.create(DEVICE_NOTIFICATION_PATH_QNAME, "stream-path").intern();
  75     private static final NodeIdentifier DATASTORE_NODEID = NodeIdentifier.create(DATASTORE_QNAME);
  76     private static final NodeIdentifier SCOPE_NODEID = NodeIdentifier.create(SCOPE_QNAME);
  77     private static final NodeIdentifier OUTPUT_TYPE_NODEID = NodeIdentifier.create(OUTPUT_TYPE_QNAME);
  78     private static final NodeIdentifier DEVICE_NOTIFICATION_PATH_NODEID =
  79         NodeIdentifier.create(DEVICE_NOTIFICATION_PATH_QNAME);
  80     private static final AugmentationIdentifier SAL_REMOTE_AUG_IDENTIFIER = new AugmentationIdentifier(
  81         ImmutableSet.of(SCOPE_QNAME, DATASTORE_QNAME, OUTPUT_TYPE_QNAME));
  82
  83     private CreateStreamUtil() {
  84         // Hidden on purpose
  85     }
  86
  87     /**
  88      * Create data-change-event or notification stream with POST operation via RPC.
  89      *
  90      * @param payload      Input of RPC - example in JSON (data-change-event stream):
  91      *                     <pre>
  92      *                     {@code
  93      *                         {
  94      *                             "input": {
  95      *                                 "path": "/toaster:toaster/toaster:toasterStatus",
  96      *                                 "sal-remote-augment:datastore": "OPERATIONAL",
  97      *                                 "sal-remote-augment:scope": "ONE"
  98      *                             }
  99      *                         }
 100      *                     }
 101      *                     </pre>
 102      * @param refSchemaCtx Reference to {@link EffectiveModelContext}.
 103      * @return {@link DOMRpcResult} - Output of RPC - example in JSON:
 104      *     <pre>
 105      *     {@code
 106      *         {
 107      *             "output": {
 108      *                 "stream-name": "toaster:toaster/toaster:toasterStatus/datastore=OPERATIONAL/scope=ONE"
 109      *             }
 110      *         }
 111      *     }
 112      *     </pre>
 113      */
 114     static DOMRpcResult createDataChangeNotifiStream(final NormalizedNodePayload payload,
 115             final EffectiveModelContext refSchemaCtx) {
 116         // parsing out of container with settings and path
 117         final ContainerNode data = (ContainerNode) requireNonNull(payload).getData();
 118         final QName qname = payload.getInstanceIdentifierContext().getSchemaNode().getQName();
 119         final YangInstanceIdentifier path = preparePath(data, qname);
 120
 121         // building of stream name
 122         final StringBuilder streamNameBuilder = new StringBuilder(
 123                 prepareDataChangeNotifiStreamName(path, requireNonNull(refSchemaCtx), data));
 124         final NotificationOutputType outputType = prepareOutputType(data);
 125         if (outputType.equals(NotificationOutputType.JSON)) {
 126             streamNameBuilder.append('/').append(outputType.getName());
 127         }
 128         final String streamName = streamNameBuilder.toString();
 129
 130         // registration of the listener
 131         ListenersBroker.getInstance().registerDataChangeListener(path, streamName, outputType);
 132
 133         // building of output
 134         final QName outputQname = QName.create(qname, "output");
 135         final QName streamNameQname = QName.create(qname, "stream-name");
 136
 137         return new DefaultDOMRpcResult(Builders.containerBuilder()
 138             .withNodeIdentifier(new NodeIdentifier(outputQname))
 139             .withChild(ImmutableNodes.leafNode(streamNameQname, streamName))
 140             .build());
 141     }
 142
 143     /**
 144      * Create device notification stream.
 145      *
 146      * @param baseUrl base Url
 147      * @param payload data
 148      * @param refSchemaCtx Reference to {@link EffectiveModelContext}.
 149      * @param streamUtil stream utility
 150      * @param mountPointService dom mount point service
 151      * @return {@link DOMRpcResult} - Output of RPC - example in JSON
 152      */
 153     static DOMRpcResult createDeviceNotificationListener(final String baseUrl, final NormalizedNodePayload payload,
 154             final EffectiveModelContext refSchemaCtx, final SubscribeToStreamUtil streamUtil,
 155             final DOMMountPointService mountPointService) {
 156         // parsing out of container with settings and path
 157         final ContainerNode data = (ContainerNode) requireNonNull(payload).getData();
 158         final YangInstanceIdentifier value =
 159             (YangInstanceIdentifier) data.findChildByArg(DEVICE_NOTIFICATION_PATH_NODEID)
 160             .map(DataContainerChild::body)
 161             .orElseThrow(() -> new RestconfDocumentedException("Mount point not available", ErrorType.APPLICATION,
 162                 ErrorTag.OPERATION_FAILED));
 163         final DOMMountPoint mountPoint = mountPointService.getMountPoint(value)
 164             .orElseThrow(() -> new RestconfDocumentedException("Mount point not available", ErrorType.APPLICATION,
 165                 ErrorTag.OPERATION_FAILED));
 166
 167         final String deviceName = extractDeviceName(value);
 168         final NotificationOutputType outputType = prepareOutputType(data);
 169         EffectiveModelContext effectiveModelContext = mountPoint.getService(DOMSchemaService.class).get()
 170             .getGlobalContext();
 171         Collection<? extends NotificationDefinition> notificationDefinitions = mountPoint.getService(
 172                 DOMSchemaService.class).get().getGlobalContext()
 173             .getNotifications();
 174         if (notificationDefinitions == null || notificationDefinitions.isEmpty()) {
 175             throw new RestconfDocumentedException("Device does not support notification", ErrorType.APPLICATION,
 176                 ErrorTag.OPERATION_FAILED);
 177         }
 178
 179         Set<Absolute> absolutes = notificationDefinitions.stream()
 180             .map(notificationDefinition -> Absolute.of(notificationDefinition.getQName()))
 181             .collect(Collectors.toUnmodifiableSet());
 182
 183         final DeviceNotificationListenerAdaptor notificationListenerAdapter = ListenersBroker.getInstance()
 184             .registerDeviceNotificationListener(deviceName, outputType, effectiveModelContext, mountPointService,
 185                 mountPoint.getIdentifier());
 186         notificationListenerAdapter.listen(mountPoint.getService(DOMNotificationService.class).get(), absolutes);
 187
 188         // building of output
 189         return new DefaultDOMRpcResult(Builders.containerBuilder()
 190             .withNodeIdentifier(new NodeIdentifier(SubscribeDeviceNotificationOutput.QNAME))
 191             .withChild(ImmutableNodes.leafNode(DEVICE_NOTIFICATION_STREAM_PATH, baseUrl + deviceName
 192                 + "?" + RestconfStreamsConstants.NOTIFICATION_TYPE + "=" + RestconfStreamsConstants.DEVICE))
 193             .build());
 194     }
 195
 196     private static String extractDeviceName(final YangInstanceIdentifier iid) {
 197         return ((YangInstanceIdentifier.NodeIdentifierWithPredicates.Singleton)iid.getLastPathArgument())
 198             .values().getElement().toString();
 199     }
 200
 201     /**
 202      * Prepare {@link NotificationOutputType}.
 203      *
 204      * @param data Container with stream settings (RPC create-stream).
 205      * @return Parsed {@link NotificationOutputType}.
 206      */
 207     private static NotificationOutputType prepareOutputType(final ContainerNode data) {
 208         final String outputName = extractStringLeaf(data, OUTPUT_TYPE_NODEID);
 209         return outputName != null ? NotificationOutputType.valueOf(outputName) : NotificationOutputType.XML;
 210     }
 211
 212     /**
 213      * Prepare stream name.
 214      *
 215      * @param path          Path of element from which data-change-event notifications are going to be generated.
 216      * @param schemaContext Schema context.
 217      * @param data          Container with stream settings (RPC create-stream).
 218      * @return Parsed stream name.
 219      */
 220     private static String prepareDataChangeNotifiStreamName(final YangInstanceIdentifier path,
 221             final EffectiveModelContext schemaContext, final ContainerNode data) {
 222         final String datastoreName = extractStringLeaf(data, DATASTORE_NODEID);
 223         final LogicalDatastoreType datastoreType = datastoreName != null ? LogicalDatastoreType.valueOf(datastoreName)
 224             : LogicalDatastoreType.CONFIGURATION;
 225
 226         final String scopeName = extractStringLeaf(data, SCOPE_NODEID);
 227         // FIXME: this is not really used
 228         final Scope scope = scopeName != null ? Scope.ofName(scopeName) : Scope.BASE;
 229
 230         return RestconfStreamsConstants.DATA_SUBSCRIPTION
 231             + "/" + ListenersBroker.createStreamNameFromUri(IdentifierCodec.serialize(path, schemaContext)
 232                 + "/" + RestconfStreamsConstants.DATASTORE_PARAM_NAME + "=" + datastoreType
 233                 + "/" + RestconfStreamsConstants.SCOPE_PARAM_NAME + "=" + scope);
 234     }
 235
 236     /**
 237      * Prepare {@link YangInstanceIdentifier} of stream source.
 238      *
 239      * @param data          Container with stream settings (RPC create-stream).
 240      * @param qualifiedName QName of the input RPC context (used only in debugging).
 241      * @return Parsed {@link YangInstanceIdentifier} of data element from which the data-change-event notifications
 242      *     are going to be generated.
 243      */
 244     private static YangInstanceIdentifier preparePath(final ContainerNode data, final QName qualifiedName) {
 245         final Object pathValue = data.findChildByArg(new NodeIdentifier(QName.create(qualifiedName, "path")))
 246             .map(DataContainerChild::body)
 247             .orElse(null);
 248         if (!(pathValue instanceof YangInstanceIdentifier)) {
 249             LOG.debug("Instance identifier {} was not normalized correctly", qualifiedName);
 250             throw new RestconfDocumentedException(
 251                     "Instance identifier was not normalized correctly",
 252                     ErrorType.APPLICATION,
 253                     ErrorTag.OPERATION_FAILED);
 254         }
 255         return (YangInstanceIdentifier) pathValue;
 256     }
 257
 258     private static @Nullable String extractStringLeaf(final ContainerNode data, final NodeIdentifier childName) {
 259         final DataContainerChild augNode = data.childByArg(SAL_REMOTE_AUG_IDENTIFIER);
 260         if (augNode instanceof AugmentationNode) {
 261             final DataContainerChild enumNode = ((AugmentationNode) augNode).childByArg(childName);
 262             if (enumNode instanceof LeafNode) {
 263                 final Object value = enumNode.body();
 264                 if (value instanceof String) {
 265                     return (String) value;
 266                 }
 267             }
 268         }
 269         return null;
 270     }
 271
 272     /**
 273      * Create YANG notification stream using notification definition in YANG schema.
 274      *
 275      * @param notificationDefinition YANG notification definition.
 276      * @param refSchemaCtx           Reference to {@link EffectiveModelContext}
 277      * @param outputType             Output type (XML or JSON).
 278      * @return {@link NotificationListenerAdapter}
 279      */
 280     static NotificationListenerAdapter createYangNotifiStream(final NotificationDefinition notificationDefinition,
 281             final EffectiveModelContext refSchemaCtx, final NotificationOutputType outputType) {
 282         final String streamName = parseNotificationStreamName(requireNonNull(notificationDefinition),
 283                 requireNonNull(refSchemaCtx), requireNonNull(outputType.getName()));
 284         final Optional<NotificationListenerAdapter> listenerForStreamName = ListenersBroker.getInstance()
 285                 .getNotificationListenerFor(streamName);
 286         return listenerForStreamName.orElseGet(() -> ListenersBroker.getInstance().registerNotificationListener(
 287                 Absolute.of(notificationDefinition.getQName()), streamName, outputType));
 288     }
 289
 290     private static String parseNotificationStreamName(final NotificationDefinition notificationDefinition,
 291             final EffectiveModelContext refSchemaCtx, final String outputType) {
 292         final QName notificationDefinitionQName = notificationDefinition.getQName();
 293         final Module module = refSchemaCtx.findModule(
 294                 notificationDefinitionQName.getModule().getNamespace(),
 295                 notificationDefinitionQName.getModule().getRevision()).orElse(null);
 296         requireNonNull(module, String.format("Module for namespace %s does not exist.",
 297                 notificationDefinitionQName.getModule().getNamespace()));
 298
 299         final StringBuilder streamNameBuilder = new StringBuilder();
 300         streamNameBuilder.append(RestconfStreamsConstants.NOTIFICATION_STREAM)
 301                 .append('/')
 302                 .append(module.getName())
 303                 .append(':')
 304                 .append(notificationDefinitionQName.getLocalName());
 305         if (outputType.equals(NotificationOutputType.JSON.getName())) {
 306             streamNameBuilder.append('/').append(NotificationOutputType.JSON.getName());
 307         }
 308         return streamNameBuilder.toString();
 309     }
 310 }