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.restconf.common.context.InstanceIdentifierContext;
  36 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  37 import org.opendaylight.restconf.common.errors.RestconfError.ErrorTag;
  38 import org.opendaylight.restconf.common.errors.RestconfError.ErrorType;
  39 import org.opendaylight.restconf.common.util.DataChangeScope;
  40 import org.opendaylight.restconf.nb.rfc8040.Rfc8040.MonitoringModule;
  41 import org.opendaylight.restconf.nb.rfc8040.handlers.NotificationServiceHandler;
  42 import org.opendaylight.restconf.nb.rfc8040.handlers.SchemaContextHandler;
  43 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
  44 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
  45 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
  46 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
  47 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  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.yangtools.concepts.ListenerRegistration;
  53 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  56 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  57 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  58 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  59 import org.opendaylight.yangtools.yang.model.api.Module;
  60 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  61 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  62 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 /**
  67  * Subscribe to stream util class.
  68  */
  69 public final class SubscribeToStreamUtil {
  70
  71     private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
  72     private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
  73             .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
  74             .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
  75             .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
  76             .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
  77             .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
  78             .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
  79             .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
  80             .appendOffset("+HH:MM", "Z").toFormatter();
  81
  82     private SubscribeToStreamUtil() {
  83         throw new UnsupportedOperationException("Utility class");
  84     }
  85
  86     /**
  87      * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
  88      * about listener to DS according to ietf-restconf-monitoring.
  89      *
  90      * @param identifier              Name of the stream.
  91      * @param uriInfo                 URI information.
  92      * @param notificationQueryParams Query parameters of notification.
  93      * @param handlersHolder          Holder of handlers for notifications.
  94      * @return Stream location for listening.
  95      */
  96     @SuppressWarnings("rawtypes")
  97     public static URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
  98             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
  99         final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
 100         if (Strings.isNullOrEmpty(streamName)) {
 101             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 102         }
 103         final Optional<NotificationListenerAdapter> notificationListenerAdapter =
 104                 ListenersBroker.getInstance().getNotificationListenerFor(streamName);
 105
 106         if (!notificationListenerAdapter.isPresent()) {
 107             throw new RestconfDocumentedException(String.format(
 108                     "Stream with name %s was not found.", streamName),
 109                     ErrorType.PROTOCOL,
 110                     ErrorTag.UNKNOWN_ELEMENT);
 111         }
 112
 113         final DOMDataTreeReadWriteTransaction writeTransaction = handlersHolder
 114                 .getTransactionChainHandler()
 115                 .get()
 116                 .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         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 DOMDataTreeReadWriteTransaction writeTransaction
 212                 = handlersHolder.getTransactionChainHandler().get().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         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     @SuppressWarnings("rawtypes")
 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 RestconfStreamsConstants.SCHEMA_UPGRADE_SECURED_URI:
 293                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_SECURED_URI);
 294                 break;
 295             case RestconfStreamsConstants.SCHEMA_UPGRADE_URI:
 296             default:
 297                 uriBuilder.scheme(RestconfStreamsConstants.SCHEMA_SUBSCRIBE_URI);
 298         }
 299         return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + RestconfConstants.SLASH + streamName)
 300                 .build();
 301     }
 302
 303     /**
 304      * Register data change listener in DOM data broker and set it to listener on stream.
 305      *
 306      * @param datastore     {@link LogicalDatastoreType}
 307      * @param listener      listener on specific stream
 308      * @param domDataBroker data broker for register data change listener
 309      */
 310     private static void registration(final LogicalDatastoreType datastore, final ListenerAdapter listener,
 311             final DOMDataBroker domDataBroker) {
 312         if (listener.isListening()) {
 313             return;
 314         }
 315
 316         final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
 317                 .getInstance(DOMDataTreeChangeService.class);
 318         if (changeService == null) {
 319             throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 320         }
 321
 322         final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(datastore, listener.getPath());
 323         final ListenerRegistration<ListenerAdapter> registration =
 324                 changeService.registerDataTreeChangeListener(root, listener);
 325         listener.setRegistration(registration);
 326     }
 327
 328     static boolean checkExist(final SchemaContext schemaContext,
 329                               final DOMDataTreeReadOperations readWriteTransaction) {
 330         boolean exist;
 331         try {
 332             return readWriteTransaction.exists(LogicalDatastoreType.OPERATIONAL,
 333                     IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).get();
 334         } catch (final InterruptedException | ExecutionException exception) {
 335             throw new RestconfDocumentedException("Problem while checking data if exists", exception);
 336         }
 337     }
 338
 339     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 340             final NotificationServiceHandler notificationServiceHandler) {
 341         if (listener.isListening()) {
 342             return;
 343         }
 344
 345         final SchemaPath path = listener.getSchemaPath();
 346         final ListenerRegistration<DOMNotificationListener> registration =
 347                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 348         listener.setRegistration(registration);
 349     }
 350
 351     /**
 352      * Parse out enumeration from URI.
 353      *
 354      * @param clazz Target enumeration type.
 355      * @param value String representation of enumeration value.
 356      * @return Parsed enumeration type.
 357      */
 358     private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
 359         if (value == null || value.equals("")) {
 360             return null;
 361         }
 362         return ResolveEnumUtil.resolveEnum(clazz, value);
 363     }
 364
 365 }