restconf/sal-rest-connector/src/main/java/org/opendaylight/restconf/restful/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.restful.utils;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.base.Strings;
  12 import java.net.URI;
  13 import java.text.DateFormat;
  14 import java.text.ParseException;
  15 import java.text.SimpleDateFormat;
  16 import java.util.ArrayList;
  17 import java.util.Date;
  18 import java.util.HashMap;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Map.Entry;
  22 import javax.ws.rs.core.UriBuilder;
  23 import javax.ws.rs.core.UriInfo;
  24 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  25 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  26 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  27 import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
  28 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
  29 import org.opendaylight.netconf.sal.restconf.impl.InstanceIdentifierContext;
  30 import org.opendaylight.netconf.sal.restconf.impl.RestconfDocumentedException;
  31 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorTag;
  32 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorType;
  33 import org.opendaylight.netconf.sal.streams.listeners.ListenerAdapter;
  34 import org.opendaylight.netconf.sal.streams.listeners.NotificationListenerAdapter;
  35 import org.opendaylight.netconf.sal.streams.listeners.Notificator;
  36 import org.opendaylight.netconf.sal.streams.websockets.WebSocketServer;
  37 import org.opendaylight.restconf.handlers.DOMDataBrokerHandler;
  38 import org.opendaylight.restconf.handlers.NotificationServiceHandler;
  39 import org.opendaylight.restconf.handlers.SchemaContextHandler;
  40 import org.opendaylight.restconf.utils.RestconfConstants;
  41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  42 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  43 import org.opendaylight.yangtools.yang.common.QName;
  44 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  45 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  46 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  47 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  48 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  49 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  50 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 /**
  55  * Subscribe to stream util class
  56  *
  57  */
  58 public final class SubscribeToStreamUtil {
  59
  60     private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
  61
  62     private SubscribeToStreamUtil() {
  63         throw new UnsupportedOperationException("Util class");
  64     }
  65
  66     /**
  67      * Parse enum from URI
  68      *
  69      * @param clazz
  70      *            - enum type
  71      * @param value
  72      *            - string of enum value
  73      * @return enum
  74      */
  75     public static <T> T parseURIEnum(final Class<T> clazz, final String value) {
  76         if ((value == null) || value.equals("")) {
  77             return null;
  78         }
  79         return StreamUtil.resolveEnum(clazz, value);
  80     }
  81
  82     /**
  83      * Prepare map of values from URI
  84      *
  85      * @param identifier
  86      *            - URI
  87      * @return {@link Map}
  88      */
  89     public static Map<String, String> mapValuesFromUri(final String identifier) {
  90         final HashMap<String, String> result = new HashMap<>();
  91         final String[] tokens = identifier.split(String.valueOf(RestconfConstants.SLASH));
  92         for (final String token : tokens) {
  93             final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
  94             if (paramToken.length == 2) {
  95                 result.put(paramToken[0], paramToken[1]);
  96             }
  97         }
  98         return result;
  99     }
 100
 101     /**
 102      * Register data change listener in dom data broker and set it to listener
 103      * on stream
 104      *
 105      * @param ds
 106      *            - {@link LogicalDatastoreType}
 107      * @param scope
 108      *            - {@link DataChangeScope}
 109      * @param listener
 110      *            - listener on specific stream
 111      * @param domDataBroker
 112      *            - data broker for register data change listener
 113      */
 114     private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
 115             final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
 116         if (listener.isListening()) {
 117             return;
 118         }
 119
 120         final YangInstanceIdentifier path = listener.getPath();
 121         final ListenerRegistration<DOMDataChangeListener> registration = domDataBroker.registerDataChangeListener(ds,
 122                 path, listener, scope);
 123
 124         listener.setRegistration(registration);
 125     }
 126
 127     /**
 128      * Get port from web socket server. If doesn't exit, create it.
 129      *
 130      * @return port
 131      */
 132     private static int prepareNotificationPort() {
 133         int port = RestconfStreamsConstants.NOTIFICATION_PORT;
 134         try {
 135             final WebSocketServer webSocketServer = WebSocketServer.getInstance();
 136             port = webSocketServer.getPort();
 137         } catch (final NullPointerException e) {
 138             WebSocketServer.createInstance(RestconfStreamsConstants.NOTIFICATION_PORT);
 139         }
 140         return port;
 141     }
 142
 143     /**
 144      * Register listeners by streamName in identifier to listen to yang notifications
 145      *
 146      * @param identifier
 147      *            - identifier as stream name
 148      * @param uriInfo
 149      *            - for getting base URI information
 150      * @param start
 151      *            - start-time query parameter
 152      * @param stop
 153      *            - stop-time query parameter
 154      * @param notifiServiceHandler
 155      *            - DOMNotificationService handler for register listeners
 156      * @return location for listening
 157      */
 158     public static URI notifStream(final String identifier, final UriInfo uriInfo, final Date start, final Date stop,
 159             final NotificationServiceHandler notifiServiceHandler) {
 160         final String streamName = Notificator.createStreamNameFromUri(identifier);
 161         if (Strings.isNullOrEmpty(streamName)) {
 162             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 163         }
 164         final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
 165         if ((listeners == null) || listeners.isEmpty()) {
 166             throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
 167                     ErrorTag.UNKNOWN_ELEMENT);
 168         }
 169
 170         for (final NotificationListenerAdapter listener : listeners) {
 171             registerToListenNotification(listener, notifiServiceHandler);
 172             listener.setTime(start, stop);
 173         }
 174
 175         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
 176         int notificationPort = RestconfStreamsConstants.NOTIFICATION_PORT;
 177         try {
 178             final WebSocketServer webSocketServerInstance = WebSocketServer.getInstance();
 179             notificationPort = webSocketServerInstance.getPort();
 180         } catch (final NullPointerException e) {
 181             WebSocketServer.createInstance(RestconfStreamsConstants.NOTIFICATION_PORT);
 182         }
 183         final UriBuilder uriToWebsocketServerBuilder = uriBuilder.port(notificationPort).scheme("ws");
 184         final URI uriToWebsocketServer = uriToWebsocketServerBuilder.replacePath(streamName).build();
 185
 186         return uriToWebsocketServer;
 187     }
 188
 189     private static void registerToListenNotification(final NotificationListenerAdapter listener,
 190             final NotificationServiceHandler notificationServiceHandler) {
 191         if (listener.isListening()) {
 192             return;
 193         }
 194
 195         final SchemaPath path = listener.getSchemaPath();
 196         final ListenerRegistration<DOMNotificationListener> registration =
 197                 notificationServiceHandler.get().registerNotificationListener(listener, path);
 198
 199         listener.setRegistration(registration);
 200     }
 201
 202     /**
 203      * Prepare InstanceIdentifierContext for Location leaf
 204      *
 205      * @param schemaHandler
 206      *            - schemaContext handler
 207      * @return InstanceIdentifier of Location leaf
 208      */
 209     public static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
 210         final QName qnameBase = QName.create("subscribe:to:notification", "2016-10-28", "notifi");
 211         final DataSchemaNode location = ((ContainerSchemaNode) schemaHandler.get()
 212                 .findModuleByNamespaceAndRevision(qnameBase.getNamespace(), qnameBase.getRevision())
 213                 .getDataChildByName(qnameBase)).getDataChildByName(QName.create(qnameBase, "location"));
 214         final List<PathArgument> path = new ArrayList<>();
 215         path.add(NodeIdentifier.create(qnameBase));
 216         path.add(NodeIdentifier.create(QName.create(qnameBase, "location")));
 217
 218         return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location, null,
 219                 schemaHandler.get());
 220     }
 221
 222     /**
 223      * Register listener by streamName in identifier to listen to yang notifications
 224      *
 225      * @param identifier
 226      *            - identifier as stream name
 227      * @param uriInfo
 228      *            - for getting base URI information
 229      * @param start
 230      *            - start-time query parameter
 231      * @param stop
 232      *            - stop-time query parameter
 233      * @param domDataBrokerHandler
 234      *            - DOMDataBroker handler for register listener
 235      * @return location for listening
 236      */
 237     public static URI dataSubs(final String identifier, final UriInfo uriInfo, final Date start, final Date stop,
 238             final DOMDataBrokerHandler domDataBrokerHandler) {
 239         final Map<String, String> mapOfValues = SubscribeToStreamUtil.mapValuesFromUri(identifier);
 240
 241         final LogicalDatastoreType ds = SubscribeToStreamUtil.parseURIEnum(LogicalDatastoreType.class,
 242                 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
 243         if (ds == null) {
 244             final String msg = "Stream name doesn't contains datastore value (pattern /datastore=)";
 245             LOG.debug(msg);
 246             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 247         }
 248
 249         final DataChangeScope scope = SubscribeToStreamUtil.parseURIEnum(DataChangeScope.class,
 250                 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
 251         if (scope == null) {
 252             final String msg = "Stream name doesn't contains datastore value (pattern /scope=)";
 253             LOG.warn(msg);
 254             throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
 255         }
 256
 257         final String streamName = Notificator.createStreamNameFromUri(identifier);
 258
 259         final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 260         Preconditions.checkNotNull(listener, "Listener doesn't exist : " + streamName);
 261
 262         listener.setTimer(start, stop);
 263
 264         SubscribeToStreamUtil.registration(ds, scope, listener, domDataBrokerHandler.get());
 265
 266         final int port = SubscribeToStreamUtil.prepareNotificationPort();
 267
 268         final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
 269         final UriBuilder uriToWebSocketServer =
 270                 uriBuilder.port(port).scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
 271         return uriToWebSocketServer.replacePath(streamName).build();
 272     }
 273
 274     public static Date parseDateFromQueryParam(final Entry<String, List<String>> entry) {
 275         final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
 276         String numOf_ms = "";
 277         final String value = event.getValue();
 278         if (value.contains(".")) {
 279             numOf_ms = numOf_ms + ".";
 280             final int lastChar = value.contains("Z") ? value.indexOf("Z") : (value.contains("+") ? value.indexOf("+")
 281                     : (value.contains("-") ? value.indexOf("-") : value.length()));
 282             for (int i = 0; i < (lastChar - value.indexOf(".") - 1); i++) {
 283                 numOf_ms = numOf_ms + "S";
 284             }
 285         }
 286         String zone = "";
 287         if (!value.contains("Z")) {
 288             zone = zone + "XXX";
 289         }
 290         final DateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss" + numOf_ms + zone);
 291
 292         try {
 293             return dateFormatter.parse(value.contains("Z") ? value.replace('T', ' ').substring(0, value.indexOf("Z"))
 294                     : value.replace('T', ' '));
 295         } catch (final ParseException e) {
 296             throw new RestconfDocumentedException("Cannot parse of value in date: " + value + e);
 297         }
 298     }
 299 }