restconf/sal-rest-connector/src/main/java/org/opendaylight/restconf/restful/utils/SubscribeToStreamUtil.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.restful.utils;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.base.Strings;
  12 import java.net.URI;
  13 import java.time.Instant;
  14 import java.time.format.DateTimeFormatter;
  15 import java.time.format.DateTimeFormatterBuilder;
  16 import java.time.format.DateTimeParseException;
  17 import java.time.temporal.ChronoField;
  18 import java.time.temporal.TemporalAccessor;
  19 import java.util.ArrayList;
  20 import java.util.HashMap;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Map.Entry;
  24 import javax.ws.rs.core.UriBuilder;
  25 import javax.ws.rs.core.UriInfo;
  26 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  29 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  30 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  31 import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
  32 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  33 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
  34 import org.opendaylight.netconf.sal.restconf.impl.InstanceIdentifierContext;
  35 import org.opendaylight.netconf.sal.restconf.impl.RestconfDocumentedException;
  36 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorTag;
  37 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorType;
  38 import org.opendaylight.netconf.sal.streams.listeners.ListenerAdapter;
  39 import org.opendaylight.netconf.sal.streams.listeners.NotificationListenerAdapter;
  40 import org.opendaylight.netconf.sal.streams.listeners.Notificator;
  41 import org.opendaylight.netconf.sal.streams.websockets.WebSocketServer;
  42 import org.opendaylight.restconf.Rfc8040.MonitoringModule;
  43 import org.opendaylight.restconf.handlers.NotificationServiceHandler;
  44 import org.opendaylight.restconf.handlers.SchemaContextHandler;
  45 import org.opendaylight.restconf.parser.IdentifierCodec;
  46 import org.opendaylight.restconf.restful.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
  47 import org.opendaylight.restconf.restful.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
  48 import org.opendaylight.restconf.utils.RestconfConstants;
  49 import org.opendaylight.restconf.utils.mapping.RestconfMappingNodeUtil;
  50 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  51 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  52 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  53 import org.opendaylight.yangtools.yang.common.QName;
  54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  56 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  57 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  58 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  59 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  60 import org.opendaylight.yangtools.yang.model.api.Module;
  61 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  62 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  63 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  64 import org.slf4j.Logger;
  65 import org.slf4j.LoggerFactory;
  66
  67 /**
  68  * Subscribe to stream util class
  69  *
  70  */
  71 public final class SubscribeToStreamUtil {
  72
  73     private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
  74     private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
  75             .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
  76             .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
  77             .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
  78             .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
  79             .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
  80             .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
  81             .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
  82             .appendOffset("+HH:MM", "Z").toFormatter();
  83
  84     private SubscribeToStreamUtil() {
  85         throw new UnsupportedOperationException("Util class");
  86     }
  87
  88     /**
  89      * Register listeners by streamName in identifier to listen to yang
  90      * notifications, put or delete info about listener to DS according to
  91      * ietf-restconf-monitoring
  92      *
  93      * @param identifier
  94      *            - identifier as stream name
  95      * @param uriInfo
  96      *            - for getting base URI information
  97      * @param notificationQueryParams
  98      *            - query parameters of notification
  99      * @param handlersHolder
 100      *            - holder of handlers for notifications
 101      * @return location for listening
 102      */
 103     @SuppressWarnings("rawtypes")
 104     public static URI notifYangStream(final String identifier, final UriInfo uriInfo,
 105             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 106         final String streamName = Notificator.createStreamNameFromUri(identifier);
 107         if (Strings.isNullOrEmpty(streamName)) {
 108             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 109         }
 110         List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
 111         if (identifier.contains(RestconfConstants.SLASH + NotificationOutputType.JSON.getName())) {
 112             listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.JSON.getName());
 113         } else {
 114             listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.XML.getName());
 115         }
 116         if ((listeners == null) || listeners.isEmpty()) {
 117             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
 118                     ErrorTag.UNKNOWN_ELEMENT);
 119         }
 120
 121         final DOMDataReadWriteTransaction wTx =
 122                 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
 123         final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
 124         final boolean exist = checkExist(schemaContext, wTx);
 125
 126         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 127         for (final NotificationListenerAdapter listener : listeners) {
 128             registerToListenNotification(listener, handlersHolder.getNotificationServiceHandler());
 129             listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
 130                     notificationQueryParams.getFilter());
 131             listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 132             final NormalizedNode mapToStreams =
 133                     RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(listener.getSchemaPath().getLastComponent(),
 134                             schemaContext.getNotifications(), notificationQueryParams.getStart(),
 135                             listener.getOutputType(), uri, getMonitoringModule(schemaContext), exist);
 136             writeDataToDS(schemaContext, listener.getSchemaPath().getLastComponent().getLocalName(), wTx, exist,
 137                     mapToStreams);
 138         }
 139         submitData(wTx);
 140
 141         return uri;
 142     }
 143
 144     static List<NotificationListenerAdapter>
 145             pickSpecificListenerByOutput(final List<NotificationListenerAdapter> listeners, final String outputType) {
 146         for (final NotificationListenerAdapter notificationListenerAdapter : listeners) {
 147             if (notificationListenerAdapter.getOutputType().equals(outputType)) {
 148                 final List<NotificationListenerAdapter> list = new ArrayList<>();
 149                 list.add(notificationListenerAdapter);
 150                 return list;
 151             }
 152         }
 153         return listeners;
 154     }
 155
 156     /**
 157      * Prepare InstanceIdentifierContext for Location leaf
 158      *
 159      * @param schemaHandler
 160      *            - schemaContext handler
 161      * @return InstanceIdentifier of Location leaf
 162      */
 163     public static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
 164         final QName qnameBase = QName.create("subscribe:to:notification", "2016-10-28", "notifi");
 165         final DataSchemaNode location = ((ContainerSchemaNode) schemaHandler.get()
 166                 .findModuleByNamespaceAndRevision(qnameBase.getNamespace(), qnameBase.getRevision())
 167                 .getDataChildByName(qnameBase)).getDataChildByName(QName.create(qnameBase, "location"));
 168         final List<PathArgument> path = new ArrayList<>();
 169         path.add(NodeIdentifier.create(qnameBase));
 170         path.add(NodeIdentifier.create(QName.create(qnameBase, "location")));
 171
 172         return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location, null,
 173                 schemaHandler.get());
 174     }
 175
 176     /**
 177      * Register listener by streamName in identifier to listen to data change
 178      * notifications, put or delete info about listener to DS according to
 179      * ietf-restconf-monitoring
 180      *
 181      * @param identifier
 182      *            - identifier as stream name
 183      * @param uriInfo
 184      *            - for getting base URI information
 185      * @param notificationQueryParams
 186      *            - query parameters of notification
 187      * @param handlersHolder
 188      *            - holder of handlers for notifications
 189      * @return location for listening
 190      */
 191     @SuppressWarnings("rawtypes")
 192     public static URI notifiDataStream(final String identifier, final UriInfo uriInfo,
 193             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 194         final Map<String, String> mapOfValues = SubscribeToStreamUtil.mapValuesFromUri(identifier);
 195
 196         final LogicalDatastoreType ds = SubscribeToStreamUtil.parseURIEnum(LogicalDatastoreType.class,
 197                 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
 198         if (ds == null) {
 199             final String msg = "Stream name doesn't contains datastore value (pattern /datastore=)";
 200             LOG.debug(msg);
 201             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 202         }
 203
 204         final DataChangeScope scope = SubscribeToStreamUtil.parseURIEnum(DataChangeScope.class,
 205                 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
 206         if (scope == null) {
 207             final String msg = "Stream name doesn't contains datastore value (pattern /scope=)";
 208             LOG.warn(msg);
 209             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 210         }
 211
 212         final String streamName = Notificator.createStreamNameFromUri(identifier);
 213
 214         final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 215         Preconditions.checkNotNull(listener, "Listener doesn't exist : " + streamName);
 216
 217         listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
 218                 notificationQueryParams.getFilter());
 219         listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 220
 221         registration(ds, scope, listener, handlersHolder.getDomDataBrokerHandler().get());
 222
 223         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 224
 225         final DOMDataReadWriteTransaction wTx =
 226                 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
 227         final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
 228         final boolean exist = checkExist(schemaContext, wTx);
 229
 230         final NormalizedNode mapToStreams = RestconfMappingNodeUtil
 231                 .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.getPath(),
 232                         notificationQueryParams.getStart(), listener.getOutputType(), uri,
 233                         getMonitoringModule(schemaContext), exist, schemaContext);
 234         writeDataToDS(schemaContext, listener.getPath().getLastPathArgument().getNodeType().getLocalName(), wTx, exist,
 235                 mapToStreams);
 236         submitData(wTx);
 237         return uri;
 238     }
 239
 240     public static Module getMonitoringModule(final SchemaContext schemaContext) {
 241         final Module monitoringModule =
 242                 schemaContext.findModuleByNamespaceAndRevision(MonitoringModule.URI_MODULE, MonitoringModule.DATE);
 243         return monitoringModule;
 244     }
 245
 246     /**
 247      * Parse input of query parameters - start-time or stop-time - from
 248      * {@link DateAndTime} format to {@link Instant} format
 249      *
 250      * @param entry
 251      *            - start-time or stop-time as string in {@link DateAndTime}
 252      *            format
 253      * @return parsed {@link Instant} by entry
 254      */
 255     public static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 256         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 257         final String value = event.getValue();
 258         final TemporalAccessor p;
 259         try {
 260             p = FORMATTER.parse(value);
 261         } catch (DateTimeParseException e) {
 262             throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
 263         }
 264         return Instant.from(p);
 265
 266     }
 267
 268     @SuppressWarnings("rawtypes")
 269     static void writeDataToDS(final SchemaContext schemaContext, final String name,
 270             final DOMDataReadWriteTransaction wTx, final boolean exist, final NormalizedNode mapToStreams) {
 271         String pathId = "";
 272         if (exist) {
 273             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 274         } else {
 275             pathId = MonitoringModule.PATH_TO_STREAMS;
 276         }
 277         wTx.merge(LogicalDatastoreType.OPERATIONAL, IdentifierCodec.deserialize(pathId, schemaContext),
 278                 mapToStreams);
 279     }
 280
 281     static void submitData(final DOMDataReadWriteTransaction wTx) {
 282         try {
 283             wTx.submit().checkedGet();
 284         } catch (final TransactionCommitFailedException e) {
 285             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 286         }
 287     }
 288
 289     /**
 290      * Prepare map of values from URI
 291      *
 292      * @param identifier
 293      *            - URI
 294      * @return {@link Map}
 295      */
 296     public static Map<String, String> mapValuesFromUri(final String identifier) {
 297         final HashMap<String, String> result = new HashMap<>();
 298         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 299             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 300             if (paramToken.length == 2) {
 301                 result.put(paramToken[0], paramToken[1]);
 302             }
 303         }
 304         return result;
 305     }
 306
 307     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 308         final int port = SubscribeToStreamUtil.prepareNotificationPort();
 309
 310         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
 311         final UriBuilder uriToWebSocketServer =
 312                 uriBuilder.port(port).scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
 313         final URI uri = uriToWebSocketServer.replacePath(streamName).build();
 314         return uri;
 315     }
 316
 317     /**
 318      * Register data change listener in dom data broker and set it to listener
 319      * on stream
 320      *
 321      * @param ds
 322      *            - {@link LogicalDatastoreType}
 323      * @param scope
 324      *            - {@link DataChangeScope}
 325      * @param listener
 326      *            - listener on specific stream
 327      * @param domDataBroker
 328      *            - data broker for register data change listener
 329      */
 330     @SuppressWarnings("deprecation")
 331     private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
 332             final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
 333         if (listener.isListening()) {
 334             return;
 335         }
 336
 337         final YangInstanceIdentifier path = listener.getPath();
 338         final ListenerRegistration<DOMDataChangeListener> registration =
 339                 domDataBroker.registerDataChangeListener(ds, path, listener, scope);
 340
 341         listener.setRegistration(registration);
 342     }
 343
 344     /**
 345      * Get port from web socket server. If doesn't exit, create it.
 346      *
 347      * @return port
 348      */
 349     private static int prepareNotificationPort() {
 350         int port = RestconfStreamsConstants.NOTIFICATION_PORT;
 351         try {
 352             final WebSocketServer webSocketServer = WebSocketServer.getInstance();
 353             port = webSocketServer.getPort();
 354         } catch (final NullPointerException e) {
 355             WebSocketServer.createInstance(RestconfStreamsConstants.NOTIFICATION_PORT);
 356         }
 357         return port;
 358     }
 359
 360     static boolean checkExist(final SchemaContext schemaContext, final DOMDataReadWriteTransaction wTx) {
 361         boolean exist;
 362         try {
 363             exist = wTx.exists(LogicalDatastoreType.OPERATIONAL,
 364                     IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).checkedGet();
 365         } catch (final ReadFailedException e1) {
 366             throw new RestconfDocumentedException("Problem while checking data if exists", e1);
 367         }
 368         return exist;
 369     }
 370
 371     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 372             final NotificationServiceHandler notificationServiceHandler) {
 373         if (listener.isListening()) {
 374             return;
 375         }
 376
 377         final SchemaPath path = listener.getSchemaPath();
 378         final ListenerRegistration<DOMNotificationListener> registration =
 379                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 380
 381         listener.setRegistration(registration);
 382     }
 383
 384     /**
 385      * Parse enum from URI
 386      *
 387      * @param clazz
 388      *            - enum type
 389      * @param value
 390      *            - string of enum value
 391      * @return enum
 392      */
 393     private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
 394         if ((value == null) || value.equals("")) {
 395             return null;
 396         }
 397         return ResolveEnumUtil.resolveEnum(clazz, value);
 398     }
 399
 400 }