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     @SuppressWarnings("rawtypes")
  99     public 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 = RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
 131                 notificationListenerAdapter.get().getSchemaPath().getLastComponent(),
 132                 schemaContext.getNotifications(), notificationQueryParams.getStart(),
 133                 notificationListenerAdapter.get().getOutputType(), uri, getMonitoringModule(schemaContext),
 134                 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     public 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     @SuppressWarnings("rawtypes")
 178     public static URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
 179             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 180         final Map<String, String> mapOfValues = mapValuesFromUri(identifier);
 181         final LogicalDatastoreType datastoreType = parseURIEnum(
 182                 LogicalDatastoreType.class,
 183                 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
 184         if (datastoreType == null) {
 185             final String message = "Stream name doesn't contain datastore value (pattern /datastore=)";
 186             LOG.debug(message);
 187             throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 188         }
 189
 190         final DataChangeScope scope = parseURIEnum(
 191                 DataChangeScope.class,
 192                 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
 193         if (scope == null) {
 194             final String message = "Stream name doesn't contains datastore value (pattern /scope=)";
 195             LOG.warn(message);
 196             throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 197         }
 198
 199         final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
 200         final Optional<ListenerAdapter> listener = ListenersBroker.getInstance().getDataChangeListenerFor(streamName);
 201         Preconditions.checkArgument(listener.isPresent(), "Listener doesn't exist : " + streamName);
 202
 203         listener.get().setQueryParams(
 204                 notificationQueryParams.getStart(),
 205                 notificationQueryParams.getStop().orElse(null),
 206                 notificationQueryParams.getFilter().orElse(null),
 207                 false);
 208         listener.get().setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
 209         registration(datastoreType, listener.get(), handlersHolder.getDomDataBrokerHandler().get());
 210
 211         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 212         final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
 213         final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
 214         final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
 215         final boolean exist = checkExist(schemaContext, writeTransaction);
 216
 217         final NormalizedNode mapToStreams = RestconfMappingNodeUtil
 218                 .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.get().getPath(),
 219                         notificationQueryParams.getStart(), listener.get().getOutputType(), uri,
 220                         getMonitoringModule(schemaContext), exist, schemaContext);
 221         writeDataToDS(schemaContext, listener.get().getPath().getLastPathArgument().getNodeType().getLocalName(),
 222                 writeTransaction, exist, mapToStreams);
 223         submitData(writeTransaction);
 224         transactionChain.close();
 225         return uri;
 226     }
 227
 228     static Module getMonitoringModule(final SchemaContext schemaContext) {
 229         return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
 230     }
 231
 232     /**
 233      * Parse input of query parameters - start-time or stop-time - from {@link DateAndTime} format
 234      * to {@link Instant} format.
 235      *
 236      * @param entry Start-time or stop-time as string in {@link DateAndTime} format.
 237      * @return Parsed {@link Instant} by entry.
 238      */
 239     public static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 240         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 241         final String value = event.getValue();
 242         final TemporalAccessor accessor;
 243         try {
 244             accessor = FORMATTER.parse(value);
 245         } catch (final DateTimeParseException e) {
 246             throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
 247         }
 248         return Instant.from(accessor);
 249     }
 250
 251     @SuppressWarnings("rawtypes")
 252     static void writeDataToDS(final SchemaContext schemaContext, final String name,
 253             final DOMDataTreeReadWriteTransaction readWriteTransaction, final boolean exist,
 254             final NormalizedNode mapToStreams) {
 255         String pathId;
 256         if (exist) {
 257             pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
 258         } else {
 259             pathId = MonitoringModule.PATH_TO_STREAMS;
 260         }
 261         readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL,
 262                 IdentifierCodec.deserialize(pathId, schemaContext), mapToStreams);
 263     }
 264
 265     static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
 266         try {
 267             readWriteTransaction.commit().get();
 268         } catch (final InterruptedException | ExecutionException e) {
 269             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 270         }
 271     }
 272
 273     /**
 274      * Prepare map of URI parameter-values.
 275      *
 276      * @param identifier String identification of URI.
 277      * @return Map od URI parameters and values.
 278      */
 279     private static Map<String, String> mapValuesFromUri(final String identifier) {
 280         final HashMap<String, String> result = new HashMap<>();
 281         for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
 282             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
 283             if (paramToken.length == 2) {
 284                 result.put(paramToken[0], paramToken[1]);
 285             }
 286         }
 287         return result;
 288     }
 289
 290     static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
 291         final String scheme = uriInfo.getAbsolutePath().getScheme();
 292         final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
 293         switch (scheme) {
 294             case RestconfStreamsConstants.SCHEMA_UPGRADE_SECURED_URI:
 295                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_SECURED_URI);
 296                 break;
 297             case RestconfStreamsConstants.SCHEMA_UPGRADE_URI:
 298             default:
 299                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_URI);
 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
 367 }