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.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 public final class SubscribeToStreamUtil {
  71
  72     private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
  73     private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
  74             .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
  75             .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
  76             .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
  77             .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
  78             .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
  79             .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
  80             .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
  81             .appendOffset("+HH:MM", "Z").toFormatter();
  82
  83     private SubscribeToStreamUtil() {
  84         throw new UnsupportedOperationException("Utility class");
  85     }
  86
  87     /**
  88      * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
  89      * about listener to DS according to ietf-restconf-monitoring.
  90      *
  91      * @param identifier              Name of the stream.
  92      * @param uriInfo                 URI information.
  93      * @param notificationQueryParams Query parameters of notification.
  94      * @param handlersHolder          Holder of handlers for notifications.
  95      * @return Stream location for listening.
  96      */
  97     @SuppressWarnings("rawtypes")
  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 = RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
 130                 notificationListenerAdapter.get().getSchemaPath().getLastComponent(),
 131                 schemaContext.getNotifications(), notificationQueryParams.getStart(),
 132                 notificationListenerAdapter.get().getOutputType(), uri, getMonitoringModule(schemaContext),
 133                 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     @SuppressWarnings("rawtypes")
 177     public 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 SchemaContext 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     public 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     @SuppressWarnings("rawtypes")
 251     static void writeDataToDS(final SchemaContext schemaContext, final String name,
 252             final DOMDataTreeReadWriteTransaction readWriteTransaction, final boolean exist,
 253             final NormalizedNode mapToStreams) {
 254         String pathId;
 255         if (exist) {
 256             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 257         } else {
 258             pathId = MonitoringModule.PATH_TO_STREAMS;
 259         }
 260         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL,
 261                 IdentifierCodec.deserialize(pathId, schemaContext), mapToStreams);
 262     }
 263
 264     static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
 265         try {
 266             readWriteTransaction.commit().get();
 267         } catch (final InterruptedException | ExecutionException e) {
 268             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 269         }
 270     }
 271
 272     /**
 273      * Prepare map of URI parameter-values.
 274      *
 275      * @param identifier String identification of URI.
 276      * @return Map od URI parameters and values.
 277      */
 278     private static Map<String, String> mapValuesFromUri(final String identifier) {
 279         final HashMap<String, String> result = new HashMap<>();
 280         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 281             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 282             if (paramToken.length == 2) {
 283                 result.put(paramToken[0], paramToken[1]);
 284             }
 285         }
 286         return result;
 287     }
 288
 289     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 290         final String scheme = uriInfo.getAbsolutePath().getScheme();
 291         final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 292         switch (scheme) {
 293             case RestconfStreamsConstants.SCHEMA_UPGRADE_SECURED_URI:
 294                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_SECURED_URI);
 295                 break;
 296             case RestconfStreamsConstants.SCHEMA_UPGRADE_URI:
 297             default:
 298                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_URI);
 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 }