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