2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.restconf.restful.utils;
10 import com.google.common.base.Preconditions;
11 import com.google.common.base.Strings;
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;
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.common.api.data.ReadFailedException;
27 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
28 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
29 import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
30 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
31 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
32 import org.opendaylight.netconf.sal.restconf.impl.InstanceIdentifierContext;
33 import org.opendaylight.netconf.sal.restconf.impl.RestconfDocumentedException;
34 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorTag;
35 import org.opendaylight.netconf.sal.restconf.impl.RestconfError.ErrorType;
36 import org.opendaylight.netconf.sal.streams.listeners.ListenerAdapter;
37 import org.opendaylight.netconf.sal.streams.listeners.NotificationListenerAdapter;
38 import org.opendaylight.netconf.sal.streams.listeners.Notificator;
39 import org.opendaylight.netconf.sal.streams.websockets.WebSocketServer;
40 import org.opendaylight.restconf.Rfc8040.MonitoringModule;
41 import org.opendaylight.restconf.handlers.NotificationServiceHandler;
42 import org.opendaylight.restconf.handlers.SchemaContextHandler;
43 import org.opendaylight.restconf.parser.IdentifierCodec;
44 import org.opendaylight.restconf.restful.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
45 import org.opendaylight.restconf.restful.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
46 import org.opendaylight.restconf.utils.RestconfConstants;
47 import org.opendaylight.restconf.utils.mapping.RestconfMappingNodeUtil;
48 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
49 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
50 import org.opendaylight.yangtools.concepts.ListenerRegistration;
51 import org.opendaylight.yangtools.yang.common.QName;
52 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
53 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
55 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
56 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
57 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
58 import org.opendaylight.yangtools.yang.model.api.Module;
59 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
60 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
61 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
62 import org.slf4j.Logger;
63 import org.slf4j.LoggerFactory;
66 * Subscribe to stream util class
69 public final class SubscribeToStreamUtil {
71 private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
73 private SubscribeToStreamUtil() {
74 throw new UnsupportedOperationException("Util class");
78 * Register listeners by streamName in identifier to listen to yang
79 * notifications, put or delete info about listener to DS according to
80 * ietf-restconf-monitoring
83 * - identifier as stream name
85 * - for getting base URI information
86 * @param notificationQueryParams
87 * - query parameters of notification
88 * @param handlersHolder
89 * - holder of handlers for notifications
90 * @return location for listening
92 @SuppressWarnings("rawtypes")
93 public static URI notifYangStream(final String identifier, final UriInfo uriInfo,
94 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
95 final String streamName = Notificator.createStreamNameFromUri(identifier);
96 if (Strings.isNullOrEmpty(streamName)) {
97 throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
99 List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
100 if (identifier.contains(RestconfConstants.SLASH + NotificationOutputType.JSON.getName())) {
101 listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.JSON.getName());
103 listeners = pickSpecificListenerByOutput(listeners, NotificationOutputType.XML.getName());
105 if ((listeners == null) || listeners.isEmpty()) {
106 throw new RestconfDocumentedException("Stream was not found.", ErrorType.PROTOCOL,
107 ErrorTag.UNKNOWN_ELEMENT);
110 final DOMDataReadWriteTransaction wTx =
111 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
112 final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
113 final boolean exist = checkExist(schemaContext, wTx);
115 final URI uri = prepareUriByStreamName(uriInfo, streamName);
116 for (final NotificationListenerAdapter listener : listeners) {
117 registerToListenNotification(listener, handlersHolder.getNotificationServiceHandler());
118 listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
119 notificationQueryParams.getFilter());
120 listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
121 final NormalizedNode mapToStreams =
122 RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(listener.getSchemaPath().getLastComponent(),
123 schemaContext.getNotifications(), notificationQueryParams.getStart(),
124 listener.getOutputType(), uri, getMonitoringModule(schemaContext), exist);
125 writeDataToDS(schemaContext, listener.getSchemaPath().getLastComponent().getLocalName(), wTx, exist,
133 static List<NotificationListenerAdapter>
134 pickSpecificListenerByOutput(final List<NotificationListenerAdapter> listeners, final String outputType) {
135 for (final NotificationListenerAdapter notificationListenerAdapter : listeners) {
136 if (notificationListenerAdapter.getOutputType().equals(outputType)) {
137 final List<NotificationListenerAdapter> list = new ArrayList<>();
138 list.add(notificationListenerAdapter);
146 * Prepare InstanceIdentifierContext for Location leaf
148 * @param schemaHandler
149 * - schemaContext handler
150 * @return InstanceIdentifier of Location leaf
152 public static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
153 final QName qnameBase = QName.create("subscribe:to:notification", "2016-10-28", "notifi");
154 final DataSchemaNode location = ((ContainerSchemaNode) schemaHandler.get()
155 .findModuleByNamespaceAndRevision(qnameBase.getNamespace(), qnameBase.getRevision())
156 .getDataChildByName(qnameBase)).getDataChildByName(QName.create(qnameBase, "location"));
157 final List<PathArgument> path = new ArrayList<>();
158 path.add(NodeIdentifier.create(qnameBase));
159 path.add(NodeIdentifier.create(QName.create(qnameBase, "location")));
161 return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location, null,
162 schemaHandler.get());
166 * Register listener by streamName in identifier to listen to data change
167 * notifications, put or delete info about listener to DS according to
168 * ietf-restconf-monitoring
171 * - identifier as stream name
173 * - for getting base URI information
174 * @param notificationQueryParams
175 * - query parameters of notification
176 * @param handlersHolder
177 * - holder of handlers for notifications
178 * @return location for listening
180 @SuppressWarnings("rawtypes")
181 public static URI notifiDataStream(final String identifier, final UriInfo uriInfo,
182 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
183 final Map<String, String> mapOfValues = SubscribeToStreamUtil.mapValuesFromUri(identifier);
185 final LogicalDatastoreType ds = SubscribeToStreamUtil.parseURIEnum(LogicalDatastoreType.class,
186 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
188 final String msg = "Stream name doesn't contains datastore value (pattern /datastore=)";
190 throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
193 final DataChangeScope scope = SubscribeToStreamUtil.parseURIEnum(DataChangeScope.class,
194 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
196 final String msg = "Stream name doesn't contains datastore value (pattern /scope=)";
198 throw new RestconfDocumentedException(msg, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
201 final String streamName = Notificator.createStreamNameFromUri(identifier);
203 final ListenerAdapter listener = Notificator.getListenerFor(streamName);
204 Preconditions.checkNotNull(listener, "Listener doesn't exist : " + streamName);
206 listener.setQueryParams(notificationQueryParams.getStart(), notificationQueryParams.getStop(),
207 notificationQueryParams.getFilter());
208 listener.setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
210 registration(ds, scope, listener, handlersHolder.getDomDataBrokerHandler().get());
212 final URI uri = prepareUriByStreamName(uriInfo, streamName);
214 final DOMDataReadWriteTransaction wTx =
215 handlersHolder.getTransactionChainHandler().get().newReadWriteTransaction();
216 final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
217 final boolean exist = checkExist(schemaContext, wTx);
219 final NormalizedNode mapToStreams = RestconfMappingNodeUtil
220 .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.getPath(),
221 notificationQueryParams.getStart(), listener.getOutputType(), uri,
222 getMonitoringModule(schemaContext), exist, schemaContext);
223 writeDataToDS(schemaContext, listener.getPath().getLastPathArgument().getNodeType().getLocalName(), wTx, exist,
229 public static Module getMonitoringModule(final SchemaContext schemaContext) {
230 final Module monitoringModule =
231 schemaContext.findModuleByNamespaceAndRevision(MonitoringModule.URI_MODULE, MonitoringModule.DATE);
232 return monitoringModule;
236 * Parse input of query parameters - start-time or stop-time - from
237 * {@link DateAndTime} format to {@link Date} format
240 * - start-time or stop-time as string in {@link DateAndTime}
242 * @return parsed {@link Date} by entry
244 public static Date parseDateFromQueryParam(final Entry<String, List<String>> entry) {
245 final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
246 String numOf_ms = "";
247 final String value = event.getValue();
248 if (value.contains(".")) {
249 numOf_ms = numOf_ms + ".";
250 final int lastChar = value.contains("Z") ? value.indexOf("Z") : (value.contains("+") ? value.indexOf("+")
251 : (value.contains("-") ? value.indexOf("-") : value.length()));
252 for (int i = 0; i < (lastChar - value.indexOf(".") - 1); i++) {
253 numOf_ms = numOf_ms + "S";
257 if (!value.contains("Z")) {
260 final DateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss" + numOf_ms + zone);
263 return dateFormatter.parse(value.contains("Z") ? value.replace('T', ' ').substring(0, value.indexOf("Z"))
264 : value.replace('T', ' '));
265 } catch (final ParseException e) {
266 throw new RestconfDocumentedException("Cannot parse of value in date: " + value + e);
270 @SuppressWarnings("rawtypes")
271 static void writeDataToDS(final SchemaContext schemaContext, final String name,
272 final DOMDataReadWriteTransaction wTx, final boolean exist, final NormalizedNode mapToStreams) {
275 pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
277 pathId = MonitoringModule.PATH_TO_STREAMS;
279 wTx.merge(LogicalDatastoreType.OPERATIONAL, IdentifierCodec.deserialize(pathId, schemaContext),
283 static void submitData(final DOMDataReadWriteTransaction wTx) {
285 wTx.submit().checkedGet();
286 } catch (final TransactionCommitFailedException e) {
287 throw new RestconfDocumentedException("Problem while putting data to DS.", e);
292 * Prepare map of values from URI
296 * @return {@link Map}
298 public static Map<String, String> mapValuesFromUri(final String identifier) {
299 final HashMap<String, String> result = new HashMap<>();
300 final String[] tokens = identifier.split(String.valueOf(RestconfConstants.SLASH));
301 for (final String token : tokens) {
302 final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
303 if (paramToken.length == 2) {
304 result.put(paramToken[0], paramToken[1]);
310 static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
311 final int port = SubscribeToStreamUtil.prepareNotificationPort();
313 final UriBuilder uriBuilder = uriInfo.getAbsolutePathBuilder();
314 final UriBuilder uriToWebSocketServer =
315 uriBuilder.port(port).scheme(RestconfStreamsConstants.SCHEMA_SUBSCIBRE_URI);
316 final URI uri = uriToWebSocketServer.replacePath(streamName).build();
321 * Register data change listener in dom data broker and set it to listener
325 * - {@link LogicalDatastoreType}
327 * - {@link DataChangeScope}
329 * - listener on specific stream
330 * @param domDataBroker
331 * - data broker for register data change listener
333 @SuppressWarnings("deprecation")
334 private static void registration(final LogicalDatastoreType ds, final DataChangeScope scope,
335 final ListenerAdapter listener, final DOMDataBroker domDataBroker) {
336 if (listener.isListening()) {
340 final YangInstanceIdentifier path = listener.getPath();
341 final ListenerRegistration<DOMDataChangeListener> registration =
342 domDataBroker.registerDataChangeListener(ds, path, listener, scope);
344 listener.setRegistration(registration);
348 * Get port from web socket server. If doesn't exit, create it.
352 private static int prepareNotificationPort() {
353 int port = RestconfStreamsConstants.NOTIFICATION_PORT;
355 final WebSocketServer webSocketServer = WebSocketServer.getInstance();
356 port = webSocketServer.getPort();
357 } catch (final NullPointerException e) {
358 WebSocketServer.createInstance(RestconfStreamsConstants.NOTIFICATION_PORT);
363 static boolean checkExist(final SchemaContext schemaContext, final DOMDataReadWriteTransaction wTx) {
366 exist = wTx.exists(LogicalDatastoreType.OPERATIONAL,
367 IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).checkedGet();
368 } catch (final ReadFailedException e1) {
369 throw new RestconfDocumentedException("Problem while checking data if exists", e1);
374 private static void registerToListenNotification(final NotificationListenerAdapter listener,
375 final NotificationServiceHandler notificationServiceHandler) {
376 if (listener.isListening()) {
380 final SchemaPath path = listener.getSchemaPath();
381 final ListenerRegistration<DOMNotificationListener> registration =
382 notificationServiceHandler.get().registerNotificationListener(listener, path);
384 listener.setRegistration(registration);
388 * Parse enum from URI
393 * - string of enum value
396 private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
397 if ((value == null) || value.equals("")) {
400 return ResolveEnumUtil.resolveEnum(clazz, value);