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