restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/rests/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.nb.rfc8040.rests.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 java.util.concurrent.ExecutionException;
  25 import javax.ws.rs.core.UriBuilder;
  26 import javax.ws.rs.core.UriInfo;
  27 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  28 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  29 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
  30 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  31 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  32 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  33 import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
  34 import org.opendaylight.restconf.common.context.InstanceIdentifierContext;
  35 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  36 import org.opendaylight.restconf.common.errors.RestconfError.ErrorTag;
  37 import org.opendaylight.restconf.common.errors.RestconfError.ErrorType;
  38 import org.opendaylight.restconf.common.util.DataChangeScope;
  39 import org.opendaylight.restconf.nb.rfc8040.Rfc8040.MonitoringModule;
  40 import org.opendaylight.restconf.nb.rfc8040.handlers.NotificationServiceHandler;
  41 import org.opendaylight.restconf.nb.rfc8040.handlers.SchemaContextHandler;
  42 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
  43 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
  44 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
  45 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  46 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.Notificator;
  47 import org.opendaylight.restconf.nb.rfc8040.streams.websockets.WebSocketServer;
  48 import org.opendaylight.restconf.nb.rfc8040.utils.RestconfConstants;
  49 import org.opendaylight.restconf.nb.rfc8040.utils.mapping.RestconfMappingNodeUtil;
  50 import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
  51 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  52 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  53 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  54 import org.opendaylight.yangtools.yang.common.QName;
  55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  56 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  57 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  58 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  59 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  60 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  61 import org.opendaylight.yangtools.yang.model.api.Module;
  62 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  63 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  64 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  65 import org.slf4j.Logger;
  66 import org.slf4j.LoggerFactory;
  67
  68 /**
  69  * Subscribe to stream util class.
  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.isEmpty()) {
 117             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
 118                     ErrorTag.UNKNOWN_ELEMENT);
 119         }
 120
 121         final DOMDataTreeReadWriteTransaction 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                 .findModule(qnameBase.getModule()).get()
 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 DOMDataTreeReadWriteTransaction 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         return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
 242     }
 243
 244     /**
 245      * Parse input of query parameters - start-time or stop-time - from
 246      * {@link DateAndTime} format to {@link Instant} format.
 247      *
 248      * @param entry
 249      *             start-time or stop-time as string in {@link DateAndTime}
 250      *            format
 251      * @return parsed {@link Instant} by entry
 252      */
 253     public static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 254         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 255         final String value = event.getValue();
 256         final TemporalAccessor p;
 257         try {
 258             p = FORMATTER.parse(value);
 259         } catch (final DateTimeParseException e) {
 260             throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
 261         }
 262         return Instant.from(p);
 263
 264     }
 265
 266     @SuppressWarnings("rawtypes")
 267     static void writeDataToDS(final SchemaContext schemaContext,
 268                               final String name, final DOMDataTreeReadWriteTransaction readWriteTransaction,
 269                               final boolean exist, final NormalizedNode mapToStreams) {
 270         String pathId = "";
 271         if (exist) {
 272             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 273         } else {
 274             pathId = MonitoringModule.PATH_TO_STREAMS;
 275         }
 276         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL, IdentifierCodec.deserialize(pathId, schemaContext),
 277                 mapToStreams);
 278     }
 279
 280     static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
 281         try {
 282             readWriteTransaction.commit().get();
 283         } catch (final InterruptedException | ExecutionException e) {
 284             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 285         }
 286     }
 287
 288     /**
 289      * Prepare map of values from URI.
 290      *
 291      * @param identifier
 292      *             URI
 293      * @return {@link Map}
 294      */
 295     public static Map<String, String> mapValuesFromUri(final String identifier) {
 296         final HashMap<String, String> result = new HashMap<>();
 297         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 298             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 299             if (paramToken.length == 2) {
 300                 result.put(paramToken[0], paramToken[1]);
 301             }
 302         }
 303         return result;
 304     }
 305
 306     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 307         final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 308
 309         prepareNotificationPort(uriInfo.getBaseUri().getPort());
 310         uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
 311         return uriBuilder.replacePath(streamName).build();
 312     }
 313
 314     /**
 315      * Register data change listener in dom data broker and set it to listener
 316      * on stream.
 317      *
 318      * @param ds
 319      *             {@link LogicalDatastoreType}
 320      * @param scope
 321      *             {@link DataChangeScope}
 322      * @param listener
 323      *             listener on specific stream
 324      * @param domDataBroker
 325      *             data broker for register data change listener
 326      */
 327     private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
 328             final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
 329         if (listener.isListening()) {
 330             return;
 331         }
 332
 333         final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
 334                 .getInstance(DOMDataTreeChangeService.class);
 335         if (changeService == null) {
 336             throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 337         }
 338
 339         final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(ds, listener.getPath());
 340         final ListenerRegistration<ListenerAdapter> registration =
 341                                     changeService.registerDataTreeChangeListener(root, listener);
 342
 343         listener.setRegistration(registration);
 344     }
 345
 346     /**
 347      * Get port from web socket server. If doesn't exit, create it.
 348      *
 349      * @param port
 350      *            - port
 351      */
 352     private static void prepareNotificationPort(final int port) {
 353         try {
 354             WebSocketServer.getInstance();
 355         } catch (final NullPointerException e) {
 356             WebSocketServer.createInstance(port);
 357         }
 358     }
 359
 360     static boolean checkExist(final SchemaContext schemaContext,
 361                               final DOMDataTreeReadTransaction readWriteTransaction) {
 362         boolean exist;
 363         try {
 364             exist = readWriteTransaction.exists(LogicalDatastoreType.OPERATIONAL,
 365                     IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).get();
 366         } catch (final InterruptedException | ExecutionException e) {
 367             throw new RestconfDocumentedException("Problem while checking data if exists", e);
 368         }
 369         return exist;
 370     }
 371
 372     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 373             final NotificationServiceHandler notificationServiceHandler) {
 374         if (listener.isListening()) {
 375             return;
 376         }
 377
 378         final SchemaPath path = listener.getSchemaPath();
 379         final ListenerRegistration<DOMNotificationListener> registration =
 380                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 381
 382         listener.setRegistration(registration);
 383     }
 384
 385     /**
 386      * Parse enum from URI.
 387      *
 388      * @param clazz
 389      *             enum type
 390      * @param value
 391      *             string of enum value
 392      * @return enum
 393      */
 394     private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
 395         if (value == null || value.equals("")) {
 396             return null;
 397         }
 398         return ResolveEnumUtil.resolveEnum(clazz, value);
 399     }
 400
 401 }