restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/rests/services/impl/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.services.impl;
   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.Optional;
  25 import java.util.concurrent.ExecutionException;
  26 import javax.ws.rs.core.UriBuilder;
  27 import javax.ws.rs.core.UriInfo;
  28 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  29 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  30 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
  31 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  32 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadOperations;
  33 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  34 import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
  35 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
  36 import org.opendaylight.restconf.common.context.InstanceIdentifierContext;
  37 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  38 import org.opendaylight.restconf.common.errors.RestconfError.ErrorTag;
  39 import org.opendaylight.restconf.common.errors.RestconfError.ErrorType;
  40 import org.opendaylight.restconf.common.util.DataChangeScope;
  41 import org.opendaylight.restconf.nb.rfc8040.Rfc8040.MonitoringModule;
  42 import org.opendaylight.restconf.nb.rfc8040.handlers.NotificationServiceHandler;
  43 import org.opendaylight.restconf.nb.rfc8040.handlers.SchemaContextHandler;
  44 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
  45 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
  46 import org.opendaylight.restconf.nb.rfc8040.rests.utils.ResolveEnumUtil;
  47 import org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants;
  48 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
  49 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
  50 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  51 import org.opendaylight.restconf.nb.rfc8040.utils.RestconfConstants;
  52 import org.opendaylight.restconf.nb.rfc8040.utils.mapping.RestconfMappingNodeUtil;
  53 import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
  54 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  55 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  56 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  57 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  58 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  59 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  60 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  61 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  62 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  63 import org.opendaylight.yangtools.yang.model.api.Module;
  64 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  65 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  66 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  67 import org.slf4j.Logger;
  68 import org.slf4j.LoggerFactory;
  69
  70 /**
  71  * Subscribe to stream util class.
  72  */
  73 final class SubscribeToStreamUtil {
  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("Utility class");
  87     }
  88
  89     /**
  90      * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
  91      * about listener to DS according to ietf-restconf-monitoring.
  92      *
  93      * @param identifier              Name of the stream.
  94      * @param uriInfo                 URI information.
  95      * @param notificationQueryParams Query parameters of notification.
  96      * @param handlersHolder          Holder of handlers for notifications.
  97      * @return Stream location for listening.
  98      */
  99     static URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
 100             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 101         final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
 102         if (Strings.isNullOrEmpty(streamName)) {
 103             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 104         }
 105         final Optional<NotificationListenerAdapter> notificationListenerAdapter =
 106                 ListenersBroker.getInstance().getNotificationListenerFor(streamName);
 107
 108         if (!notificationListenerAdapter.isPresent()) {
 109             throw new RestconfDocumentedException(String.format(
 110                     "Stream with name %s was not found.", streamName),
 111                     ErrorType.PROTOCOL,
 112                     ErrorTag.UNKNOWN_ELEMENT);
 113         }
 114
 115         final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
 116         final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
 117         final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
 118         final boolean exist = checkExist(schemaContext, writeTransaction);
 119         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 120
 121         registerToListenNotification(
 122                 notificationListenerAdapter.get(), handlersHolder.getNotificationServiceHandler());
 123         notificationListenerAdapter.get().setQueryParams(
 124                 notificationQueryParams.getStart(),
 125                 notificationQueryParams.getStop().orElse(null),
 126                 notificationQueryParams.getFilter().orElse(null),
 127                 false);
 128         notificationListenerAdapter.get().setCloseVars(
 129                 handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 130         final NormalizedNode<?, ?> mapToStreams =
 131                 RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
 132                     notificationListenerAdapter.get().getSchemaPath().getLastComponent(),
 133                     schemaContext.getNotifications(), notificationQueryParams.getStart(),
 134                     notificationListenerAdapter.get().getOutputType(), uri, getMonitoringModule(schemaContext), exist);
 135         writeDataToDS(schemaContext,
 136                 notificationListenerAdapter.get().getSchemaPath().getLastComponent().getLocalName(), writeTransaction,
 137                 exist, mapToStreams);
 138         submitData(writeTransaction);
 139         transactionChain.close();
 140         return uri;
 141     }
 142
 143     /**
 144      * Prepare InstanceIdentifierContext for Location leaf.
 145      *
 146      * @param schemaHandler Schema context handler.
 147      * @return InstanceIdentifier of Location leaf.
 148      */
 149     static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
 150         final Optional<Module> module = schemaHandler.get()
 151                 .findModule(RestconfStreamsConstants.NOTIFI_QNAME.getModule());
 152         Preconditions.checkState(module.isPresent());
 153         final Optional<DataSchemaNode> notify = module.get()
 154                 .findDataChildByName(RestconfStreamsConstants.NOTIFI_QNAME);
 155         Preconditions.checkState(notify.isPresent());
 156         final Optional<DataSchemaNode> location = ((ContainerSchemaNode) notify.get())
 157                 .findDataChildByName(RestconfStreamsConstants.LOCATION_QNAME);
 158         Preconditions.checkState(location.isPresent());
 159
 160         final List<PathArgument> path = new ArrayList<>();
 161         path.add(NodeIdentifier.create(RestconfStreamsConstants.NOTIFI_QNAME));
 162         path.add(NodeIdentifier.create(RestconfStreamsConstants.LOCATION_QNAME));
 163         return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location.get(),
 164                 null, schemaHandler.get());
 165     }
 166
 167     /**
 168      * Register listener by streamName in identifier to listen to data change notifications, and put or delete
 169      * information about listener to DS according to ietf-restconf-monitoring.
 170      *
 171      * @param identifier              Identifier as stream name.
 172      * @param uriInfo                 Base URI information.
 173      * @param notificationQueryParams Query parameters of notification.
 174      * @param handlersHolder          Holder of handlers for notifications.
 175      * @return Location for listening.
 176      */
 177     static URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
 178             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 179         final Map<String, String> mapOfValues = mapValuesFromUri(identifier);
 180         final LogicalDatastoreType datastoreType = parseURIEnum(
 181                 LogicalDatastoreType.class,
 182                 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
 183         if (datastoreType == null) {
 184             final String message = "Stream name doesn't contain datastore value (pattern /datastore=)";
 185             LOG.debug(message);
 186             throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 187         }
 188
 189         final DataChangeScope scope = parseURIEnum(
 190                 DataChangeScope.class,
 191                 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
 192         if (scope == null) {
 193             final String message = "Stream name doesn't contains datastore value (pattern /scope=)";
 194             LOG.warn(message);
 195             throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 196         }
 197
 198         final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
 199         final Optional<ListenerAdapter> listener = ListenersBroker.getInstance().getDataChangeListenerFor(streamName);
 200         Preconditions.checkArgument(listener.isPresent(), "Listener doesn't exist : " + streamName);
 201
 202         listener.get().setQueryParams(
 203                 notificationQueryParams.getStart(),
 204                 notificationQueryParams.getStop().orElse(null),
 205                 notificationQueryParams.getFilter().orElse(null),
 206                 false);
 207         listener.get().setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 208         registration(datastoreType, listener.get(), handlersHolder.getDomDataBrokerHandler().get());
 209
 210         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 211         final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
 212         final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
 213         final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
 214         final boolean exist = checkExist(schemaContext, writeTransaction);
 215
 216         final NormalizedNode<?, ?> mapToStreams = RestconfMappingNodeUtil
 217                 .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.get().getPath(),
 218                         notificationQueryParams.getStart(), listener.get().getOutputType(), uri,
 219                         getMonitoringModule(schemaContext), exist, schemaContext);
 220         writeDataToDS(schemaContext, listener.get().getPath().getLastPathArgument().getNodeType().getLocalName(),
 221                 writeTransaction, exist, mapToStreams);
 222         submitData(writeTransaction);
 223         transactionChain.close();
 224         return uri;
 225     }
 226
 227     static Module getMonitoringModule(final SchemaContext schemaContext) {
 228         return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
 229     }
 230
 231     /**
 232      * Parse input of query parameters - start-time or stop-time - from {@link DateAndTime} format
 233      * to {@link Instant} format.
 234      *
 235      * @param entry Start-time or stop-time as string in {@link DateAndTime} format.
 236      * @return Parsed {@link Instant} by entry.
 237      */
 238     static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 239         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 240         final String value = event.getValue();
 241         final TemporalAccessor accessor;
 242         try {
 243             accessor = FORMATTER.parse(value);
 244         } catch (final DateTimeParseException e) {
 245             throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
 246         }
 247         return Instant.from(accessor);
 248     }
 249
 250     static void writeDataToDS(final SchemaContext schemaContext, final String name,
 251             final DOMDataTreeReadWriteTransaction readWriteTransaction, final boolean exist,
 252             final NormalizedNode<?, ?> mapToStreams) {
 253         String pathId;
 254         if (exist) {
 255             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 256         } else {
 257             pathId = MonitoringModule.PATH_TO_STREAMS;
 258         }
 259         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL,
 260                 IdentifierCodec.deserialize(pathId, schemaContext), mapToStreams);
 261     }
 262
 263     static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
 264         try {
 265             readWriteTransaction.commit().get();
 266         } catch (final InterruptedException | ExecutionException e) {
 267             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 268         }
 269     }
 270
 271     /**
 272      * Prepare map of URI parameter-values.
 273      *
 274      * @param identifier String identification of URI.
 275      * @return Map od URI parameters and values.
 276      */
 277     private static Map<String, String> mapValuesFromUri(final String identifier) {
 278         final HashMap<String, String> result = new HashMap<>();
 279         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 280             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 281             if (paramToken.length == 2) {
 282                 result.put(paramToken[0], paramToken[1]);
 283             }
 284         }
 285         return result;
 286     }
 287
 288     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 289         final String scheme = uriInfo.getAbsolutePath().getScheme();
 290         final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 291         switch (scheme) {
 292             case "https":
 293                 // Secured HTTP goes to Secured WebSockets
 294                 uriBuilder.scheme("wss");
 295                 break;
 296             case "http":
 297             default:
 298                 // Unsecured HTTP and others go to unsecured WebSockets
 299                 uriBuilder.scheme("ws");
 300         }
 301         return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + RestconfConstants.SLASH + streamName)
 302                 .build();
 303     }
 304
 305     /**
 306      * Register data change listener in DOM data broker and set it to listener on stream.
 307      *
 308      * @param datastore     {@link LogicalDatastoreType}
 309      * @param listener      listener on specific stream
 310      * @param domDataBroker data broker for register data change listener
 311      */
 312     private static void registration(final LogicalDatastoreType datastore, final ListenerAdapter listener,
 313             final DOMDataBroker domDataBroker) {
 314         if (listener.isListening()) {
 315             return;
 316         }
 317
 318         final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
 319                 .getInstance(DOMDataTreeChangeService.class);
 320         if (changeService == null) {
 321             throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 322         }
 323
 324         final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(datastore, listener.getPath());
 325         final ListenerRegistration<ListenerAdapter> registration =
 326                 changeService.registerDataTreeChangeListener(root, listener);
 327         listener.setRegistration(registration);
 328     }
 329
 330     static boolean checkExist(final SchemaContext schemaContext,
 331                               final DOMDataTreeReadOperations readWriteTransaction) {
 332         boolean exist;
 333         try {
 334             return readWriteTransaction.exists(LogicalDatastoreType.OPERATIONAL,
 335                     IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).get();
 336         } catch (final InterruptedException | ExecutionException exception) {
 337             throw new RestconfDocumentedException("Problem while checking data if exists", exception);
 338         }
 339     }
 340
 341     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 342             final NotificationServiceHandler notificationServiceHandler) {
 343         if (listener.isListening()) {
 344             return;
 345         }
 346
 347         final SchemaPath path = listener.getSchemaPath();
 348         final ListenerRegistration<DOMNotificationListener> registration =
 349                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 350         listener.setRegistration(registration);
 351     }
 352
 353     /**
 354      * Parse out enumeration from URI.
 355      *
 356      * @param clazz Target enumeration type.
 357      * @param value String representation of enumeration value.
 358      * @return Parsed enumeration type.
 359      */
 360     private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
 361         if (value == null || value.equals("")) {
 362             return null;
 363         }
 364         return ResolveEnumUtil.resolveEnum(clazz, value);
 365     }
 366 }