restconf/restconf-nb-bierman02/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(), false);
 131             listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 132             final NormalizedNode mapToStreams = RestconfMappingNodeUtil
 133                     .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(), false);
 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,
 270                               final String name, final DOMDataReadWriteTransaction readWriteTransaction,
 271                               final boolean exist, final NormalizedNode mapToStreams) {
 272         String pathId = "";
 273         if (exist) {
 274             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 275         } else {
 276             pathId = MonitoringModule.PATH_TO_STREAMS;
 277         }
 278         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL, IdentifierCodec.deserialize(pathId, schemaContext),
 279                 mapToStreams);
 280     }
 281
 282     static void submitData(final DOMDataReadWriteTransaction readWriteTransaction) {
 283         try {
 284             readWriteTransaction.submit().checkedGet();
 285         } catch (final TransactionCommitFailedException e) {
 286             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 287         }
 288     }
 289
 290     /**
 291      * Prepare map of values from URI.
 292      *
 293      * @param identifier
 294      *             URI
 295      * @return {@link Map}
 296      */
 297     public static Map<String, String> mapValuesFromUri(final String identifier) {
 298         final HashMap<String, String> result = new HashMap<>();
 299         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 300             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 301             if (paramToken.length == 2) {
 302                 result.put(paramToken[0], paramToken[1]);
 303             }
 304         }
 305         return result;
 306     }
 307
 308     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 309         final int port = SubscribeToStreamUtil.prepareNotificationPort();
 310
 311         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
 312         final UriBuilder uriToWebSocketServer =
 313                 uriBuilder.port(port).scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
 314         final URI uri = uriToWebSocketServer.replacePath(streamName).build();
 315         return uri;
 316     }
 317
 318     /**
 319      * Register data change listener in dom data broker and set it to listener
 320      * on stream.
 321      *
 322      * @param ds
 323      *             {@link LogicalDatastoreType}
 324      * @param scope
 325      *             {@link DataChangeScope}
 326      * @param listener
 327      *             listener on specific stream
 328      * @param domDataBroker
 329      *             data broker for register data change listener
 330      */
 331     @SuppressWarnings("deprecation")
 332     private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
 333             final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
 334         if (listener.isListening()) {
 335             return;
 336         }
 337
 338         final YangInstanceIdentifier path = listener.getPath();
 339         final ListenerRegistration<DOMDataChangeListener> registration =
 340                 domDataBroker.registerDataChangeListener(ds, path, listener, scope);
 341
 342         listener.setRegistration(registration);
 343     }
 344
 345     /**
 346      * Get port from web socket server. If doesn't exit, create it.
 347      *
 348      * @return port
 349      */
 350     private static int prepareNotificationPort() {
 351         int port = RestconfStreamsConstants.NOTIFICATION_PORT;
 352         try {
 353             final WebSocketServer webSocketServer = WebSocketServer.getInstance();
 354             port = webSocketServer.getPort();
 355         } catch (final NullPointerException e) {
 356             WebSocketServer.createInstance(RestconfStreamsConstants.NOTIFICATION_PORT);
 357         }
 358         return port;
 359     }
 360
 361     static boolean checkExist(final SchemaContext schemaContext,
 362                               final DOMDataReadWriteTransaction readWriteTransaction) {
 363         boolean exist;
 364         try {
 365             exist = readWriteTransaction.exists(LogicalDatastoreType.OPERATIONAL,
 366                     IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).checkedGet();
 367         } catch (final ReadFailedException e1) {
 368             throw new RestconfDocumentedException("Problem while checking data if exists", e1);
 369         }
 370         return exist;
 371     }
 372
 373     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 374             final NotificationServiceHandler notificationServiceHandler) {
 375         if (listener.isListening()) {
 376             return;
 377         }
 378
 379         final SchemaPath path = listener.getSchemaPath();
 380         final ListenerRegistration<DOMNotificationListener> registration =
 381                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 382
 383         listener.setRegistration(registration);
 384     }
 385
 386     /**
 387      * Parse enum from URI.
 388      *
 389      * @param clazz
 390      *             enum type
 391      * @param value
 392      *             string of enum value
 393      * @return enum
 394      */
 395     private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
 396         if ((value == null) || value.equals("")) {
 397             return null;
 398         }
 399         return ResolveEnumUtil.resolveEnum(clazz, value);
 400     }
 401
 402 }