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