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.nb.rfc8040.rests.services.impl;
10 import com.google.common.base.Preconditions;
11 import com.google.common.base.Strings;
13 import java.util.HashMap;
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;
52 * Subscribe to stream util class.
54 abstract class SubscribeToStreamUtil {
56 * Implementation of SubscribeToStreamUtil for Server-sent events.
58 private static final class ServerSentEvents extends SubscribeToStreamUtil {
59 static final ServerSentEvents INSTANCE = new ServerSentEvents();
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();
70 * Implementation of SubscribeToStreamUtil for Web sockets.
72 private static final class WebSockets extends SubscribeToStreamUtil {
73 static final WebSockets INSTANCE = new WebSockets();
76 public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
77 final String scheme = uriInfo.getAbsolutePath().getScheme();
78 final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
81 // Secured HTTP goes to Secured WebSockets
82 uriBuilder.scheme("wss");
86 // Unsecured HTTP and others go to unsecured WebSockets
87 uriBuilder.scheme("ws");
89 return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + '/' + streamName).build();
94 private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
96 SubscribeToStreamUtil() {
100 static SubscribeToStreamUtil serverSentEvents() {
101 return ServerSentEvents.INSTANCE;
104 static SubscribeToStreamUtil webSockets() {
105 return WebSockets.INSTANCE;
109 * Prepare URL from base name and stream name.
111 * @param uriInfo base URL information
112 * @param streamName name of stream for create
115 abstract @NonNull URI prepareUriByStreamName(UriInfo uriInfo, String streamName);
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.
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.
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);
133 final Optional<NotificationListenerAdapter> notificationListenerAdapter =
134 ListenersBroker.getInstance().getNotificationListenerFor(streamName);
136 if (notificationListenerAdapter.isEmpty()) {
137 throw new RestconfDocumentedException(String.format(
138 "Stream with name %s was not found.", streamName),
140 ErrorTag.UNKNOWN_ELEMENT);
143 final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
144 final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
145 final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
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,
164 submitData(writeTransaction);
165 transactionChain.close();
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.
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.
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=)";
188 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
191 final DataChangeScope scope = parseURIEnum(
192 DataChangeScope.class,
193 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
195 final String message = "Stream name doesn't contains datastore value (pattern /scope=)";
197 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
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);
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());
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);
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();
228 static Module getMonitoringModule(final EffectiveModelContext schemaContext) {
229 return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
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),
240 private static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
242 readWriteTransaction.commit().get();
243 } catch (final InterruptedException | ExecutionException e) {
244 throw new RestconfDocumentedException("Problem while putting data to DS.", e);
249 * Prepare map of URI parameter-values.
251 * @param identifier String identification of URI.
252 * @return Map od URI parameters and values.
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]);
266 * Register data change listener in DOM data broker and set it to listener on stream.
268 * @param datastore {@link LogicalDatastoreType}
269 * @param listener listener on specific stream
270 * @param domDataBroker data broker for register data change listener
272 private static void registration(final LogicalDatastoreType datastore, final ListenerAdapter listener,
273 final DOMDataBroker domDataBroker) {
274 if (listener.isListening()) {
278 final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
279 .getInstance(DOMDataTreeChangeService.class);
280 if (changeService == null) {
281 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
284 final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(datastore, listener.getPath());
285 final ListenerRegistration<ListenerAdapter> registration =
286 changeService.registerDataTreeChangeListener(root, listener);
287 listener.setRegistration(registration);
290 private static void registerToListenNotification(final NotificationListenerAdapter listener,
291 final NotificationServiceHandler notificationServiceHandler) {
292 if (listener.isListening()) {
296 final Absolute path = listener.getSchemaPath();
297 final ListenerRegistration<DOMNotificationListener> registration =
298 notificationServiceHandler.get().registerNotificationListener(listener, path);
299 listener.setRegistration(registration);
303 * Parse out enumeration from URI.
305 * @param clazz Target enumeration type.
306 * @param value String representation of enumeration value.
307 * @return Parsed enumeration type.
309 private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
310 if (value == null || value.equals("")) {
313 return ResolveEnumUtil.resolveEnum(clazz, value);