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.DOMNotificationService;
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.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.stmt.SchemaNodeIdentifier.Absolute;
47 import org.slf4j.Logger;
48 import org.slf4j.LoggerFactory;
51 * Subscribe to stream util class.
53 abstract class SubscribeToStreamUtil {
55 * Implementation of SubscribeToStreamUtil for Server-sent events.
57 private static final class ServerSentEvents extends SubscribeToStreamUtil {
58 static final ServerSentEvents INSTANCE = new ServerSentEvents();
61 public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
62 final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
63 return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + '/'
64 + RestconfConstants.NOTIF + '/' + streamName).build();
69 * Implementation of SubscribeToStreamUtil for Web sockets.
71 private static final class WebSockets extends SubscribeToStreamUtil {
72 static final WebSockets INSTANCE = new WebSockets();
75 public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
76 final String scheme = uriInfo.getAbsolutePath().getScheme();
77 final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
80 // Secured HTTP goes to Secured WebSockets
81 uriBuilder.scheme("wss");
85 // Unsecured HTTP and others go to unsecured WebSockets
86 uriBuilder.scheme("ws");
88 return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + '/' + streamName).build();
93 private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
95 SubscribeToStreamUtil() {
99 static SubscribeToStreamUtil serverSentEvents() {
100 return ServerSentEvents.INSTANCE;
103 static SubscribeToStreamUtil webSockets() {
104 return WebSockets.INSTANCE;
108 * Prepare URL from base name and stream name.
110 * @param uriInfo base URL information
111 * @param streamName name of stream for create
114 abstract @NonNull URI prepareUriByStreamName(UriInfo uriInfo, String streamName);
117 * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
118 * about listener to DS according to ietf-restconf-monitoring.
120 * @param identifier Name of the stream.
121 * @param uriInfo URI information.
122 * @param notificationQueryParams Query parameters of notification.
123 * @param handlersHolder Holder of handlers for notifications.
124 * @return Stream location for listening.
126 final @NonNull URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
127 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
128 final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
129 if (Strings.isNullOrEmpty(streamName)) {
130 throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
133 final NotificationListenerAdapter notificationListenerAdapter = ListenersBroker.getInstance()
134 .getNotificationListenerFor(streamName)
135 .orElseThrow(() -> new RestconfDocumentedException(
136 String.format("Stream with name %s was not found.", streamName),
137 ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT));
139 final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
140 final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
141 final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
143 final URI uri = prepareUriByStreamName(uriInfo, streamName);
144 registerToListenNotification(notificationListenerAdapter, handlersHolder.getNotificationServiceHandler());
145 notificationListenerAdapter.setQueryParams(
146 notificationQueryParams.getStart(),
147 notificationQueryParams.getStop().orElse(null),
148 notificationQueryParams.getFilter().orElse(null),
149 false, notificationQueryParams.isSkipNotificationData());
150 notificationListenerAdapter.setCloseVars(
151 handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
152 final MapEntryNode mapToStreams = RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
153 notificationListenerAdapter.getSchemaPath().lastNodeIdentifier(),
154 schemaContext.getNotifications(), notificationQueryParams.getStart(),
155 notificationListenerAdapter.getOutputType(), uri);
156 writeDataToDS(schemaContext,
157 notificationListenerAdapter.getSchemaPath().lastNodeIdentifier().getLocalName(), writeTransaction,
159 submitData(writeTransaction);
160 transactionChain.close();
165 * Register listener by streamName in identifier to listen to data change notifications, and put or delete
166 * information about listener to DS according to ietf-restconf-monitoring.
168 * @param identifier Identifier as stream name.
169 * @param uriInfo Base URI information.
170 * @param notificationQueryParams Query parameters of notification.
171 * @param handlersHolder Holder of handlers for notifications.
172 * @return Location for listening.
174 final URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
175 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
176 final Map<String, String> mapOfValues = mapValuesFromUri(identifier);
177 final LogicalDatastoreType datastoreType = parseURIEnum(
178 LogicalDatastoreType.class,
179 mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
180 if (datastoreType == null) {
181 final String message = "Stream name doesn't contain datastore value (pattern /datastore=)";
183 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
186 final DataChangeScope scope = parseURIEnum(
187 DataChangeScope.class,
188 mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
190 final String message = "Stream name doesn't contains datastore value (pattern /scope=)";
192 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
195 final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
196 final Optional<ListenerAdapter> listener = ListenersBroker.getInstance().getDataChangeListenerFor(streamName);
197 Preconditions.checkArgument(listener.isPresent(), "Listener doesn't exist : " + streamName);
199 listener.get().setQueryParams(
200 notificationQueryParams.getStart(),
201 notificationQueryParams.getStop().orElse(null),
202 notificationQueryParams.getFilter().orElse(null),
203 false, notificationQueryParams.isSkipNotificationData());
204 listener.get().setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
205 registration(datastoreType, listener.get(), handlersHolder.getDataBroker());
207 final URI uri = prepareUriByStreamName(uriInfo, streamName);
208 final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
209 final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
210 final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
211 final String serializedPath = IdentifierCodec.serialize(listener.get().getPath(), schemaContext);
213 final MapEntryNode mapToStreams =
214 RestconfMappingNodeUtil.mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.get().getPath(),
215 notificationQueryParams.getStart(), listener.get().getOutputType(), uri, schemaContext, serializedPath);
216 writeDataToDS(schemaContext, serializedPath, writeTransaction, mapToStreams);
217 submitData(writeTransaction);
218 transactionChain.close();
222 private static void writeDataToDS(final EffectiveModelContext schemaContext, final String name,
223 final DOMDataTreeReadWriteTransaction readWriteTransaction, final MapEntryNode mapToStreams) {
224 readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL,
225 // FIXME: do not use IdentifierCodec here
226 IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name, schemaContext),
230 private static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
232 readWriteTransaction.commit().get();
233 } catch (final InterruptedException | ExecutionException e) {
234 throw new RestconfDocumentedException("Problem while putting data to DS.", e);
239 * Prepare map of URI parameter-values.
241 * @param identifier String identification of URI.
242 * @return Map od URI parameters and values.
244 private static Map<String, String> mapValuesFromUri(final String identifier) {
245 final HashMap<String, String> result = new HashMap<>();
246 for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
247 final String[] paramToken = token.split("=");
248 if (paramToken.length == 2) {
249 result.put(paramToken[0], paramToken[1]);
256 * Register data change listener in DOM data broker and set it to listener on stream.
258 * @param datastore {@link LogicalDatastoreType}
259 * @param listener listener on specific stream
260 * @param domDataBroker data broker for register data change listener
262 private static void registration(final LogicalDatastoreType datastore, final ListenerAdapter listener,
263 final DOMDataBroker domDataBroker) {
264 if (listener.isListening()) {
268 final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
269 .getInstance(DOMDataTreeChangeService.class);
270 if (changeService == null) {
271 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
274 final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(datastore, listener.getPath());
275 final ListenerRegistration<ListenerAdapter> registration =
276 changeService.registerDataTreeChangeListener(root, listener);
277 listener.setRegistration(registration);
280 // FIXME: this method should be in NotificationListenerAdapter
281 private static void registerToListenNotification(final NotificationListenerAdapter listener,
282 final DOMNotificationService notificationService) {
283 if (listener.isListening()) {
287 final Absolute path = listener.getSchemaPath();
288 final ListenerRegistration<DOMNotificationListener> registration =
289 notificationService.registerNotificationListener(listener, path);
290 listener.setRegistration(registration);
294 * Parse out enumeration from URI.
296 * @param clazz Target enumeration type.
297 * @param value String representation of enumeration value.
298 * @return Parsed enumeration type.
300 private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
301 if (value == null || value.equals("")) {
304 return ResolveEnumUtil.resolveEnum(clazz, value);