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 javax.ws.rs.core.UriBuilder;
  25 import javax.ws.rs.core.UriInfo;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  28 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  29 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  30 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  31 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
  32 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeIdentifier;
  33 import org.opendaylight.controller.md.sal.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  */
  72 public final class SubscribeToStreamUtil {
  73
  74     private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
  75     private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
  76             .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
  77             .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
  78             .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
  79             .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
  80             .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
  81             .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
  82             .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
  83             .appendOffset("+HH:MM", "Z").toFormatter();
  84
  85     private SubscribeToStreamUtil() {
  86         throw new UnsupportedOperationException("Util class");
  87     }
  88
  89     /**
  90      * Register listeners by streamName in identifier to listen to yang
  91      * notifications, put or delete info about listener to DS according to
  92      * ietf-restconf-monitoring.
  93      *
  94      * @param identifier
  95      *             identifier as stream name
  96      * @param uriInfo
  97      *             for getting base URI information
  98      * @param notificationQueryParams
  99      *             query parameters of notification
 100      * @param handlersHolder
 101      *             holder of handlers for notifications
 102      * @return location for listening
 103      */
 104     @SuppressWarnings("rawtypes")
 105     public static URI notifYangStream(final String identifier, final UriInfo uriInfo,
 106             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 107         final String streamName = Notificator.createStreamNameFromUri(identifier);
 108         if (Strings.isNullOrEmpty(streamName)) {
 109             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 110         }
 111         List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
 112         if (identifier.contains(RestconfConstants.SLASH + NotificationOutputType.JSON.getName())) {
 113             listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.JSON.getName());
 114         } else {
 115             listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.XML.getName());
 116         }
 117         if (listeners.isEmpty()) {
 118             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
 119                     ErrorTag.UNKNOWN_ELEMENT);
 120         }
 121
 122         final DOMDataReadWriteTransaction wTx =
 123                 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
 124         final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
 125         final boolean exist = checkExist(schemaContext, wTx);
 126
 127         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 128         for (final NotificationListenerAdapter listener : listeners) {
 129             registerToListenNotification(listener, handlersHolder.getNotificationServiceHandler());
 130             listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
 131                     notificationQueryParams.getFilter(), false);
 132             listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 133             final NormalizedNode mapToStreams = RestconfMappingNodeUtil
 134                     .mapYangNotificationStreamByIetfRestconfMonitoring(listener.getSchemaPath().getLastComponent(),
 135                             schemaContext.getNotifications(), notificationQueryParams.getStart(),
 136                             listener.getOutputType(), uri, getMonitoringModule(schemaContext), exist);
 137             writeDataToDS(schemaContext, listener.getSchemaPath().getLastComponent().getLocalName(), wTx, exist,
 138                     mapToStreams);
 139         }
 140         submitData(wTx);
 141
 142         return uri;
 143     }
 144
 145     static List<NotificationListenerAdapter>
 146             pickSpecificListenerByOutput(final List<NotificationListenerAdapter> listeners, final String outputType) {
 147         for (final NotificationListenerAdapter notificationListenerAdapter : listeners) {
 148             if (notificationListenerAdapter.getOutputType().equals(outputType)) {
 149                 final List<NotificationListenerAdapter> list = new ArrayList<>();
 150                 list.add(notificationListenerAdapter);
 151                 return list;
 152             }
 153         }
 154         return listeners;
 155     }
 156
 157     /**
 158      * Prepare InstanceIdentifierContext for Location leaf.
 159      *
 160      * @param schemaHandler
 161      *             schemaContext handler
 162      * @return InstanceIdentifier of Location leaf
 163      */
 164     public static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
 165         final QName qnameBase = QName.create("subscribe:to:notification", "2016-10-28", "notifi");
 166         final DataSchemaNode location = ((ContainerSchemaNode) schemaHandler.get()
 167                 .findModule(qnameBase.getModule()).get()
 168                 .getDataChildByName(qnameBase)).getDataChildByName(QName.create(qnameBase, "location"));
 169         final List<PathArgument> path = new ArrayList<>();
 170         path.add(NodeIdentifier.create(qnameBase));
 171         path.add(NodeIdentifier.create(QName.create(qnameBase, "location")));
 172
 173         return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location, null,
 174                 schemaHandler.get());
 175     }
 176
 177     /**
 178      * Register listener by streamName in identifier to listen to data change
 179      * notifications, put or delete info about listener to DS according to
 180      * ietf-restconf-monitoring.
 181      *
 182      * @param identifier
 183      *             identifier as stream name
 184      * @param uriInfo
 185      *             for getting base URI information
 186      * @param notificationQueryParams
 187      *             query parameters of notification
 188      * @param handlersHolder
 189      *             holder of handlers for notifications
 190      * @return location for listening
 191      */
 192     @SuppressWarnings("rawtypes")
 193     public static URI notifiDataStream(final String identifier, final UriInfo uriInfo,
 194             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 195         final Map<String, String> mapOfValues = SubscribeToStreamUtil.mapValuesFromUri(identifier);
 196
 197         final LogicalDatastoreType ds = SubscribeToStreamUtil.parseURIEnum(LogicalDatastoreType.class,
 198                 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
 199         if (ds == null) {
 200             final String msg = "Stream name doesn't contains datastore value (pattern /datastore=)";
 201             LOG.debug(msg);
 202             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 203         }
 204
 205         final DataChangeScope scope = SubscribeToStreamUtil.parseURIEnum(DataChangeScope.class,
 206                 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
 207         if (scope == null) {
 208             final String msg = "Stream name doesn't contains datastore value (pattern /scope=)";
 209             LOG.warn(msg);
 210             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 211         }
 212
 213         final String streamName = Notificator.createStreamNameFromUri(identifier);
 214
 215         final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 216         Preconditions.checkNotNull(listener, "Listener doesn't exist : " + streamName);
 217
 218         listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
 219                 notificationQueryParams.getFilter(), false);
 220         listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 221
 222         registration(ds, scope, listener, handlersHolder.getDomDataBrokerHandler().get());
 223
 224         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 225
 226         final DOMDataReadWriteTransaction wTx =
 227                 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
 228         final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
 229         final boolean exist = checkExist(schemaContext, wTx);
 230
 231         final NormalizedNode mapToStreams = RestconfMappingNodeUtil
 232                 .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.getPath(),
 233                         notificationQueryParams.getStart(), listener.getOutputType(), uri,
 234                         getMonitoringModule(schemaContext), exist, schemaContext);
 235         writeDataToDS(schemaContext, listener.getPath().getLastPathArgument().getNodeType().getLocalName(), wTx, exist,
 236                 mapToStreams);
 237         submitData(wTx);
 238         return uri;
 239     }
 240
 241     public static Module getMonitoringModule(final SchemaContext schemaContext) {
 242         return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
 243     }
 244
 245     /**
 246      * Parse input of query parameters - start-time or stop-time - from
 247      * {@link DateAndTime} format to {@link Instant} format.
 248      *
 249      * @param entry
 250      *             start-time or stop-time as string in {@link DateAndTime}
 251      *            format
 252      * @return parsed {@link Instant} by entry
 253      */
 254     public static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 255         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 256         final String value = event.getValue();
 257         final TemporalAccessor p;
 258         try {
 259             p = FORMATTER.parse(value);
 260         } catch (final DateTimeParseException e) {
 261             throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
 262         }
 263         return Instant.from(p);
 264
 265     }
 266
 267     @SuppressWarnings("rawtypes")
 268     static void writeDataToDS(final SchemaContext schemaContext,
 269                               final String name, final DOMDataReadWriteTransaction readWriteTransaction,
 270                               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         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL, IdentifierCodec.deserialize(pathId, schemaContext),
 278                 mapToStreams);
 279     }
 280
 281     static void submitData(final DOMDataReadWriteTransaction readWriteTransaction) {
 282         try {
 283             readWriteTransaction.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 UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 309
 310         prepareNotificationPort(uriInfo.getBaseUri().getPort());
 311         uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
 312         return uriBuilder.replacePath(streamName).build();
 313     }
 314
 315     /**
 316      * Register data change listener in dom data broker and set it to listener
 317      * on stream.
 318      *
 319      * @param ds
 320      *             {@link LogicalDatastoreType}
 321      * @param scope
 322      *             {@link DataChangeScope}
 323      * @param listener
 324      *             listener on specific stream
 325      * @param domDataBroker
 326      *             data broker for register data change listener
 327      */
 328     private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
 329             final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
 330         if (listener.isListening()) {
 331             return;
 332         }
 333
 334         final DOMDataTreeChangeService changeService = (DOMDataTreeChangeService)domDataBroker.getSupportedExtensions()
 335                 .get(DOMDataTreeChangeService.class);
 336         if (changeService == null) {
 337             throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 338         }
 339
 340         final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(ds, listener.getPath());
 341         final ListenerRegistration<ListenerAdapter> registration =
 342                                     changeService.registerDataTreeChangeListener(root, listener);
 343
 344         listener.setRegistration(registration);
 345     }
 346
 347     /**
 348      * Get port from web socket server. If doesn't exit, create it.
 349      *
 350      * @param port
 351      *            - port
 352      */
 353     private static void prepareNotificationPort(final int port) {
 354         try {
 355             WebSocketServer.getInstance();
 356         } catch (final NullPointerException e) {
 357             WebSocketServer.createInstance(port);
 358         }
 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 }