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 static com.google.common.base.Strings.isNullOrEmpty;
12 import com.google.common.base.Splitter;
14 import java.util.HashMap;
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.DOMDataTreeWriteOperations;
23 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
24 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
25 import org.opendaylight.restconf.nb.rfc8040.NotificationQueryParams;
26 import org.opendaylight.restconf.nb.rfc8040.URLConstants;
27 import org.opendaylight.restconf.nb.rfc8040.databind.DatabindProvider;
28 import org.opendaylight.restconf.nb.rfc8040.monitoring.RestconfStateStreams;
29 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
30 import org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants;
31 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
32 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
33 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
34 import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
35 import org.opendaylight.yangtools.yang.common.ErrorTag;
36 import org.opendaylight.yangtools.yang.common.ErrorType;
37 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
38 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
43 * Subscribe to stream util class.
45 abstract class SubscribeToStreamUtil {
47 * Implementation of SubscribeToStreamUtil for Server-sent events.
49 private static final class ServerSentEvents extends SubscribeToStreamUtil {
50 static final ServerSentEvents INSTANCE = new ServerSentEvents();
53 public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
54 return uriInfo.getBaseUriBuilder()
55 .replacePath(URLConstants.BASE_PATH + '/' + URLConstants.SSE_SUBPATH + '/' + streamName)
61 * Implementation of SubscribeToStreamUtil for Web sockets.
63 private static final class WebSockets extends SubscribeToStreamUtil {
64 static final WebSockets INSTANCE = new WebSockets();
67 public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
68 final String scheme = uriInfo.getAbsolutePath().getScheme();
69 final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
72 // Secured HTTP goes to Secured WebSockets
73 uriBuilder.scheme("wss");
77 // Unsecured HTTP and others go to unsecured WebSockets
78 uriBuilder.scheme("ws");
80 return uriBuilder.replacePath(URLConstants.BASE_PATH + '/' + streamName).build();
84 private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
85 private static final Splitter SLASH_SPLITTER = Splitter.on('/');
87 SubscribeToStreamUtil() {
91 static SubscribeToStreamUtil serverSentEvents() {
92 return ServerSentEvents.INSTANCE;
95 static SubscribeToStreamUtil webSockets() {
96 return WebSockets.INSTANCE;
100 * Prepare URL from base name and stream name.
102 * @param uriInfo base URL information
103 * @param streamName name of stream for create
106 abstract @NonNull URI prepareUriByStreamName(UriInfo uriInfo, String streamName);
109 * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
110 * about listener to DS according to ietf-restconf-monitoring.
112 * @param identifier Name of the stream.
113 * @param uriInfo URI information.
114 * @param notificationQueryParams Query parameters of notification.
115 * @param handlersHolder Holder of handlers for notifications.
116 * @return Stream location for listening.
118 final @NonNull URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
119 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
120 final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
121 if (isNullOrEmpty(streamName)) {
122 throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
125 final NotificationListenerAdapter notificationListenerAdapter = ListenersBroker.getInstance()
126 .notificationListenerFor(streamName);
127 if (notificationListenerAdapter == null) {
128 throw new RestconfDocumentedException(String.format("Stream with name %s was not found.", streamName),
129 ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT);
132 final EffectiveModelContext schemaContext = handlersHolder.getDatabindProvider().currentContext()
134 final URI uri = prepareUriByStreamName(uriInfo, streamName);
135 notificationListenerAdapter.setQueryParams(notificationQueryParams);
136 notificationListenerAdapter.listen(handlersHolder.getNotificationServiceHandler());
137 final DOMDataBroker dataBroker = handlersHolder.getDataBroker();
138 notificationListenerAdapter.setCloseVars(dataBroker, handlersHolder.getDatabindProvider());
139 final MapEntryNode mapToStreams = RestconfStateStreams.notificationStreamEntry(schemaContext,
140 notificationListenerAdapter.getSchemaPath().lastNodeIdentifier(), notificationListenerAdapter.getStart(),
141 notificationListenerAdapter.getOutputType(), uri);
143 // FIXME: how does this correlate with the transaction notificationListenerAdapter.close() will do?
144 final DOMDataTreeWriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
145 writeDataToDS(writeTransaction, mapToStreams);
146 submitData(writeTransaction);
151 * Register listener by streamName in identifier to listen to data change notifications, and put or delete
152 * information about listener to DS according to ietf-restconf-monitoring.
154 * @param identifier Identifier as stream name.
155 * @param uriInfo Base URI information.
156 * @param notificationQueryParams Query parameters of notification.
157 * @param handlersHolder Holder of handlers for notifications.
158 * @return Location for listening.
160 final URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
161 final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
162 final Map<String, String> mapOfValues = mapValuesFromUri(identifier);
164 final String datastoreParam = mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME);
165 if (isNullOrEmpty(datastoreParam)) {
166 final String message = "Stream name does not contain datastore value (pattern /datastore=)";
168 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
171 // FIXME: this is kept only for compatibility, we are not using this parameter
172 if (isNullOrEmpty(mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME))) {
173 final String message = "Stream name does not contain scope value (pattern /scope=)";
175 throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
178 final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
179 final ListenerAdapter listener = ListenersBroker.getInstance().dataChangeListenerFor(streamName);
180 if (listener == null) {
181 throw new RestconfDocumentedException("No listener found for stream " + streamName,
182 ErrorType.APPLICATION, ErrorTag.DATA_MISSING);
185 listener.setQueryParams(notificationQueryParams);
187 final DOMDataBroker dataBroker = handlersHolder.getDataBroker();
188 final DatabindProvider schemaHandler = handlersHolder.getDatabindProvider();
189 listener.setCloseVars(dataBroker, schemaHandler);
190 listener.listen(dataBroker, LogicalDatastoreType.valueOf(datastoreParam));
192 final URI uri = prepareUriByStreamName(uriInfo, streamName);
193 final EffectiveModelContext schemaContext = schemaHandler.currentContext().modelContext();
194 final String serializedPath = IdentifierCodec.serialize(listener.getPath(), schemaContext);
196 final MapEntryNode mapToStreams = RestconfStateStreams.dataChangeStreamEntry(listener.getPath(),
197 listener.getStart(), listener.getOutputType(), uri, schemaContext, serializedPath);
198 final DOMDataTreeWriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
199 writeDataToDS(writeTransaction, mapToStreams);
200 submitData(writeTransaction);
204 // FIXME: callers are utter duplicates, refactor them
205 private static void writeDataToDS(final DOMDataTreeWriteOperations tx, final MapEntryNode mapToStreams) {
206 // FIXME: use put() here
207 tx.merge(LogicalDatastoreType.OPERATIONAL, RestconfStateStreams.restconfStateStreamPath(mapToStreams.name()),
211 private static void submitData(final DOMDataTreeWriteTransaction readWriteTransaction) {
213 readWriteTransaction.commit().get();
214 } catch (final InterruptedException | ExecutionException e) {
215 throw new RestconfDocumentedException("Problem while putting data to DS.", e);
220 * Prepare map of URI parameter-values.
222 * @param identifier String identification of URI.
223 * @return Map od URI parameters and values.
225 private static Map<String, String> mapValuesFromUri(final String identifier) {
226 final var result = new HashMap<String, String>();
227 for (final String token : SLASH_SPLITTER.split(identifier)) {
228 final String[] paramToken = token.split("=");
229 if (paramToken.length == 2) {
230 result.put(paramToken[0], paramToken[1]);