restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/rests/utils/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.utils;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import java.util.Optional;
  13 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  14 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  15 import org.opendaylight.mdsal.dom.spi.DefaultDOMRpcResult;
  16 import org.opendaylight.restconf.common.context.NormalizedNodeContext;
  17 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  18 import org.opendaylight.restconf.common.errors.RestconfError.ErrorTag;
  19 import org.opendaylight.restconf.common.errors.RestconfError.ErrorType;
  20 import org.opendaylight.restconf.common.util.DataChangeScope;
  21 import org.opendaylight.restconf.nb.rfc8040.references.SchemaContextRef;
  22 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
  23 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  24 import org.opendaylight.restconf.nb.rfc8040.utils.parser.ParserIdentifier;
  25 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  26 import org.opendaylight.yangtools.yang.common.QName;
  27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  28 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  30 import org.opendaylight.yangtools.yang.data.api.schema.AugmentationNode;
  31 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  32 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  33 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  34 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  35 import org.opendaylight.yangtools.yang.model.api.Module;
  36 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
  37 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 /**
  42  * Utility class for creation of data-change-event or YANG notification streams.
  43  */
  44 public final class CreateStreamUtil {
  45
  46     private static final Logger LOG = LoggerFactory.getLogger(CreateStreamUtil.class);
  47
  48     private CreateStreamUtil() {
  49         throw new UnsupportedOperationException("Utility class");
  50     }
  51
  52     /**
  53      * Create data-change-event or notification stream with POST operation via RPC.
  54      *
  55      * @param payload      Input of RPC - example in JSON (data-change-event stream):
  56      *                     <pre>
  57      *                     {@code
  58      *                         {
  59      *                             "input": {
  60      *                                 "path": "/toaster:toaster/toaster:toasterStatus",
  61      *                                 "sal-remote-augment:datastore": "OPERATIONAL",
  62      *                                 "sal-remote-augment:scope": "ONE"
  63      *                             }
  64      *                         }
  65      *                     }
  66      *                     </pre>
  67      * @param refSchemaCtx Reference to {@link SchemaContext} - {@link SchemaContextRef}.
  68      * @return {@link DOMRpcResult} - Output of RPC - example in JSON:
  69      *     <pre>
  70      *     {@code
  71      *         {
  72      *             "output": {
  73      *                 "stream-name": "toaster:toaster/toaster:toasterStatus/datastore=OPERATIONAL/scope=ONE"
  74      *             }
  75      *         }
  76      *     }
  77      *     </pre>
  78      */
  79     public static DOMRpcResult createDataChangeNotifiStream(final NormalizedNodeContext payload,
  80             final SchemaContextRef refSchemaCtx) {
  81         // parsing out of container with settings and path
  82         final ContainerNode data = (ContainerNode) requireNonNull(payload).getData();
  83         final QName qname = payload.getInstanceIdentifierContext().getSchemaNode().getQName();
  84         final YangInstanceIdentifier path = preparePath(data, qname);
  85
  86         // building of stream name
  87         final StringBuilder streamNameBuilder = new StringBuilder(
  88                 prepareDataChangeNotifiStreamName(path, requireNonNull(refSchemaCtx).get(), data));
  89         final NotificationOutputType outputType = prepareOutputType(data);
  90         if (outputType.equals(NotificationOutputType.JSON)) {
  91             streamNameBuilder.append('/').append(outputType.getName());
  92         }
  93         final String streamName = streamNameBuilder.toString();
  94
  95         // registration of the listener
  96         ListenersBroker.getInstance().registerDataChangeListener(path, streamName, outputType);
  97
  98         // building of output
  99         final QName outputQname = QName.create(qname, RestconfStreamsConstants.OUTPUT_CONTAINER_NAME);
 100         final QName streamNameQname = QName.create(qname, RestconfStreamsConstants.OUTPUT_STREAM_NAME);
 101
 102         final ContainerNode output = ImmutableContainerNodeBuilder.create()
 103                 .withNodeIdentifier(new NodeIdentifier(outputQname))
 104                 .withChild(ImmutableNodes.leafNode(streamNameQname, streamName)).build();
 105         return new DefaultDOMRpcResult(output);
 106     }
 107
 108     /**
 109      * Prepare {@link NotificationOutputType}.
 110      *
 111      * @param data Container with stream settings (RPC create-stream).
 112      * @return Parsed {@link NotificationOutputType}.
 113      */
 114     private static NotificationOutputType prepareOutputType(final ContainerNode data) {
 115         NotificationOutputType outputType = parseEnum(
 116                 data, NotificationOutputType.class, RestconfStreamsConstants.OUTPUT_TYPE_PARAM_NAME);
 117         return outputType == null ? NotificationOutputType.XML : outputType;
 118     }
 119
 120     /**
 121      * Prepare stream name.
 122      *
 123      * @param path          Path of element from which data-change-event notifications are going to be generated.
 124      * @param schemaContext Schema context.
 125      * @param data          Container with stream settings (RPC create-stream).
 126      * @return Parsed stream name.
 127      */
 128     private static String prepareDataChangeNotifiStreamName(final YangInstanceIdentifier path,
 129             final SchemaContext schemaContext, final ContainerNode data) {
 130         LogicalDatastoreType datastoreType = parseEnum(
 131                 data, LogicalDatastoreType.class, RestconfStreamsConstants.DATASTORE_PARAM_NAME);
 132         datastoreType = datastoreType == null ? RestconfStreamsConstants.DEFAULT_DS : datastoreType;
 133
 134         DataChangeScope scope = parseEnum(data, DataChangeScope.class, RestconfStreamsConstants.SCOPE_PARAM_NAME);
 135         scope = scope == null ? RestconfStreamsConstants.DEFAULT_SCOPE : scope;
 136
 137         return RestconfStreamsConstants.DATA_SUBSCRIPTION
 138                 + "/"
 139                 + ListenersBroker.createStreamNameFromUri(
 140                 ParserIdentifier.stringFromYangInstanceIdentifier(path, schemaContext)
 141                         + RestconfStreamsConstants.DS_URI
 142                         + datastoreType
 143                         + RestconfStreamsConstants.SCOPE_URI
 144                         + scope);
 145     }
 146
 147     /**
 148      * Prepare {@link YangInstanceIdentifier} of stream source.
 149      *
 150      * @param data          Container with stream settings (RPC create-stream).
 151      * @param qualifiedName QName of the input RPC context (used only in debugging).
 152      * @return Parsed {@link YangInstanceIdentifier} of data element from which the data-change-event notifications
 153      *     are going to be generated.
 154      */
 155     private static YangInstanceIdentifier preparePath(final ContainerNode data, final QName qualifiedName) {
 156         final Optional<DataContainerChild<? extends PathArgument, ?>> path = data.getChild(
 157                 new YangInstanceIdentifier.NodeIdentifier(QName.create(
 158                         qualifiedName,
 159                         RestconfStreamsConstants.STREAM_PATH_PARAM_NAME)));
 160         Object pathValue = null;
 161         if (path.isPresent()) {
 162             pathValue = path.get().getValue();
 163         }
 164         if (!(pathValue instanceof YangInstanceIdentifier)) {
 165             LOG.debug("Instance identifier {} was not normalized correctly", qualifiedName);
 166             throw new RestconfDocumentedException(
 167                     "Instance identifier was not normalized correctly",
 168                     ErrorType.APPLICATION,
 169                     ErrorTag.OPERATION_FAILED);
 170         }
 171         return (YangInstanceIdentifier) pathValue;
 172     }
 173
 174     /**
 175      * Parsing out of enumeration from RPC create-stream body.
 176      *
 177      * @param data      Container with stream settings (RPC create-stream).
 178      * @param clazz     Enum type to be parsed out from input container.
 179      * @param paramName Local name of the enum element.
 180      * @return Parsed enumeration.
 181      */
 182     private static <T> T parseEnum(final ContainerNode data, final Class<T> clazz, final String paramName) {
 183         final Optional<DataContainerChild<? extends PathArgument, ?>> optAugNode = data.getChild(
 184                 RestconfStreamsConstants.SAL_REMOTE_AUG_IDENTIFIER);
 185         if (!optAugNode.isPresent()) {
 186             return null;
 187         }
 188         final DataContainerChild<? extends PathArgument, ?> augNode = optAugNode.get();
 189         if (!(augNode instanceof AugmentationNode)) {
 190             return null;
 191         }
 192         final Optional<DataContainerChild<? extends PathArgument, ?>> enumNode = ((AugmentationNode) augNode).getChild(
 193                 new NodeIdentifier(QName.create(RestconfStreamsConstants.SAL_REMOTE_AUGMENT, paramName)));
 194         if (!enumNode.isPresent()) {
 195             return null;
 196         }
 197         final Object value = enumNode.get().getValue();
 198         if (!(value instanceof String)) {
 199             return null;
 200         }
 201
 202         return ResolveEnumUtil.resolveEnum(clazz, (String) value);
 203     }
 204
 205     /**
 206      * Create YANG notification stream using notification definition in YANG schema.
 207      *
 208      * @param notificationDefinition YANG notification definition.
 209      * @param refSchemaCtx           Reference to {@link SchemaContext} - {@link SchemaContextRef}.
 210      * @param outputType             Output type (XML or JSON).
 211      * @return {@link NotificationListenerAdapter}
 212      */
 213     public static NotificationListenerAdapter createYangNotifiStream(
 214             final NotificationDefinition notificationDefinition, final SchemaContextRef refSchemaCtx,
 215             final NotificationOutputType outputType) {
 216         final String streamName = parseNotificationStreamName(requireNonNull(notificationDefinition),
 217                 requireNonNull(refSchemaCtx), requireNonNull(outputType.getName()));
 218         final Optional<NotificationListenerAdapter> listenerForStreamName = ListenersBroker.getInstance()
 219                 .getNotificationListenerFor(streamName);
 220         return listenerForStreamName.orElseGet(() -> ListenersBroker.getInstance().registerNotificationListener(
 221                 notificationDefinition.getPath(), streamName, outputType));
 222     }
 223
 224     private static String parseNotificationStreamName(final NotificationDefinition notificationDefinition,
 225             final SchemaContextRef refSchemaCtx, final String outputType) {
 226         final QName notificationDefinitionQName = notificationDefinition.getQName();
 227         final Module module = refSchemaCtx.findModuleByNamespaceAndRevision(
 228                 notificationDefinitionQName.getModule().getNamespace(),
 229                 notificationDefinitionQName.getModule().getRevision());
 230         requireNonNull(module, String.format("Module for namespace %s does not exist.",
 231                 notificationDefinitionQName.getModule().getNamespace()));
 232
 233         final StringBuilder streamNameBuilder = new StringBuilder();
 234         streamNameBuilder.append(RestconfStreamsConstants.CREATE_NOTIFICATION_STREAM)
 235                 .append('/')
 236                 .append(module.getName())
 237                 .append(':')
 238                 .append(notificationDefinitionQName.getLocalName());
 239         if (outputType.equals(NotificationOutputType.JSON.getName())) {
 240             streamNameBuilder.append(NotificationOutputType.JSON.getName());
 241         }
 242         return streamNameBuilder.toString();
 243     }
 244 }