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