*/
package org.opendaylight.restconf.nb.rfc8040.rests.services.impl;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
+import static com.google.common.base.Strings.isNullOrEmpty;
+
import java.net.URI;
-import java.time.Instant;
-import java.time.format.DateTimeFormatter;
-import java.time.format.DateTimeFormatterBuilder;
-import java.time.format.DateTimeParseException;
-import java.time.temporal.ChronoField;
-import java.time.temporal.TemporalAccessor;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Optional;
import java.util.concurrent.ExecutionException;
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.UriInfo;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.mdsal.dom.api.DOMDataBroker;
-import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
-import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
-import org.opendaylight.mdsal.dom.api.DOMDataTreeReadOperations;
-import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
-import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
-import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
-import org.opendaylight.restconf.common.context.InstanceIdentifierContext;
+import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteOperations;
+import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
import org.opendaylight.restconf.common.errors.RestconfError.ErrorTag;
-import org.opendaylight.restconf.common.errors.RestconfError.ErrorType;
-import org.opendaylight.restconf.common.util.DataChangeScope;
-import org.opendaylight.restconf.nb.rfc8040.Rfc8040.MonitoringModule;
-import org.opendaylight.restconf.nb.rfc8040.handlers.NotificationServiceHandler;
+import org.opendaylight.restconf.nb.rfc8040.Rfc8040;
import org.opendaylight.restconf.nb.rfc8040.handlers.SchemaContextHandler;
import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.NotificationQueryParams;
-import org.opendaylight.restconf.nb.rfc8040.rests.utils.ResolveEnumUtil;
import org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants;
import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
import org.opendaylight.restconf.nb.rfc8040.utils.RestconfConstants;
import org.opendaylight.restconf.nb.rfc8040.utils.mapping.RestconfMappingNodeUtil;
import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
-import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
+import org.opendaylight.yangtools.yang.common.ErrorType;
+import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
-import org.opendaylight.yangtools.yang.model.api.Module;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaNode;
-import org.opendaylight.yangtools.yang.model.api.SchemaPath;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Subscribe to stream util class.
*/
-final class SubscribeToStreamUtil {
+abstract class SubscribeToStreamUtil {
+ /**
+ * Implementation of SubscribeToStreamUtil for Server-sent events.
+ */
+ private static final class ServerSentEvents extends SubscribeToStreamUtil {
+ static final ServerSentEvents INSTANCE = new ServerSentEvents();
+
+ @Override
+ public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
+ final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
+ return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + '/'
+ + RestconfConstants.NOTIF + '/' + streamName).build();
+ }
+ }
+
+ /**
+ * Implementation of SubscribeToStreamUtil for Web sockets.
+ */
+ private static final class WebSockets extends SubscribeToStreamUtil {
+ static final WebSockets INSTANCE = new WebSockets();
+
+ @Override
+ public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
+ final String scheme = uriInfo.getAbsolutePath().getScheme();
+ final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
+ switch (scheme) {
+ case "https":
+ // Secured HTTP goes to Secured WebSockets
+ uriBuilder.scheme("wss");
+ break;
+ case "http":
+ default:
+ // Unsecured HTTP and others go to unsecured WebSockets
+ uriBuilder.scheme("ws");
+ }
+ return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + '/' + streamName).build();
+ }
+ }
+
+
private static final Logger LOG = LoggerFactory.getLogger(SubscribeToStreamUtil.class);
- private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
- .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
- .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
- .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
- .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
- .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
- .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
- .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
- .appendOffset("+HH:MM", "Z").toFormatter();
- private SubscribeToStreamUtil() {
- throw new UnsupportedOperationException("Utility class");
+ SubscribeToStreamUtil() {
+ // Hidden on purpose
}
+ static SubscribeToStreamUtil serverSentEvents() {
+ return ServerSentEvents.INSTANCE;
+ }
+
+ static SubscribeToStreamUtil webSockets() {
+ return WebSockets.INSTANCE;
+ }
+
+ /**
+ * Prepare URL from base name and stream name.
+ *
+ * @param uriInfo base URL information
+ * @param streamName name of stream for create
+ * @return final URL
+ */
+ abstract @NonNull URI prepareUriByStreamName(UriInfo uriInfo, String streamName);
+
/**
* Register listener by streamName in identifier to listen to yang notifications, and put or delete information
* about listener to DS according to ietf-restconf-monitoring.
* @param handlersHolder Holder of handlers for notifications.
* @return Stream location for listening.
*/
- static URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
+ final @NonNull URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
- if (Strings.isNullOrEmpty(streamName)) {
+ if (isNullOrEmpty(streamName)) {
throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
}
- final Optional<NotificationListenerAdapter> notificationListenerAdapter =
- ListenersBroker.getInstance().getNotificationListenerFor(streamName);
- if (!notificationListenerAdapter.isPresent()) {
- throw new RestconfDocumentedException(String.format(
- "Stream with name %s was not found.", streamName),
- ErrorType.PROTOCOL,
- ErrorTag.UNKNOWN_ELEMENT);
- }
+ final NotificationListenerAdapter notificationListenerAdapter = ListenersBroker.getInstance()
+ .getNotificationListenerFor(streamName)
+ .orElseThrow(() -> new RestconfDocumentedException(
+ String.format("Stream with name %s was not found.", streamName),
+ ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT));
- final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
- final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
- final SchemaContext schemaContext = handlersHolder.getSchemaHandler().get();
- final boolean exist = checkExist(schemaContext, writeTransaction);
+ final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
final URI uri = prepareUriByStreamName(uriInfo, streamName);
-
- registerToListenNotification(
- notificationListenerAdapter.get(), handlersHolder.getNotificationServiceHandler());
- notificationListenerAdapter.get().setQueryParams(
+ notificationListenerAdapter.listen(handlersHolder.getNotificationServiceHandler());
+ notificationListenerAdapter.setQueryParams(
notificationQueryParams.getStart(),
notificationQueryParams.getStop().orElse(null),
notificationQueryParams.getFilter().orElse(null),
- false);
- notificationListenerAdapter.get().setCloseVars(
- handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
- final NormalizedNode<?, ?> mapToStreams =
- RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
- notificationListenerAdapter.get().getSchemaPath().getLastComponent(),
+ false, notificationQueryParams.isSkipNotificationData());
+ final DOMDataBroker dataBroker = handlersHolder.getDataBroker();
+ notificationListenerAdapter.setCloseVars(dataBroker, handlersHolder.getSchemaHandler());
+ final MapEntryNode mapToStreams = RestconfMappingNodeUtil.mapYangNotificationStreamByIetfRestconfMonitoring(
+ notificationListenerAdapter.getSchemaPath().lastNodeIdentifier(),
schemaContext.getNotifications(), notificationQueryParams.getStart(),
- notificationListenerAdapter.get().getOutputType(), uri, getMonitoringModule(schemaContext), exist);
- writeDataToDS(schemaContext,
- notificationListenerAdapter.get().getSchemaPath().getLastComponent().getLocalName(), writeTransaction,
- exist, mapToStreams);
+ notificationListenerAdapter.getOutputType(), uri);
+
+ // FIXME: how does this correlate with the transaction notificationListenerAdapter.close() will do?
+ final DOMDataTreeWriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
+ writeDataToDS(writeTransaction, mapToStreams);
submitData(writeTransaction);
- transactionChain.close();
return uri;
}
- /**
- * Prepare InstanceIdentifierContext for Location leaf.
- *
- * @param schemaHandler Schema context handler.
- * @return InstanceIdentifier of Location leaf.
- */
- static InstanceIdentifierContext<?> prepareIIDSubsStreamOutput(final SchemaContextHandler schemaHandler) {
- final Optional<Module> module = schemaHandler.get()
- .findModule(RestconfStreamsConstants.NOTIFI_QNAME.getModule());
- Preconditions.checkState(module.isPresent());
- final Optional<DataSchemaNode> notify = module.get()
- .findDataChildByName(RestconfStreamsConstants.NOTIFI_QNAME);
- Preconditions.checkState(notify.isPresent());
- final Optional<DataSchemaNode> location = ((ContainerSchemaNode) notify.get())
- .findDataChildByName(RestconfStreamsConstants.LOCATION_QNAME);
- Preconditions.checkState(location.isPresent());
-
- final List<PathArgument> path = new ArrayList<>();
- path.add(NodeIdentifier.create(RestconfStreamsConstants.NOTIFI_QNAME));
- path.add(NodeIdentifier.create(RestconfStreamsConstants.LOCATION_QNAME));
- return new InstanceIdentifierContext<SchemaNode>(YangInstanceIdentifier.create(path), location.get(),
- null, schemaHandler.get());
- }
-
/**
* Register listener by streamName in identifier to listen to data change notifications, and put or delete
* information about listener to DS according to ietf-restconf-monitoring.
* @param handlersHolder Holder of handlers for notifications.
* @return Location for listening.
*/
- static URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
+ final URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
final Map<String, String> mapOfValues = mapValuesFromUri(identifier);
- final LogicalDatastoreType datastoreType = parseURIEnum(
- LogicalDatastoreType.class,
- mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME));
- if (datastoreType == null) {
- final String message = "Stream name doesn't contain datastore value (pattern /datastore=)";
+
+ final String datastoreParam = mapOfValues.get(RestconfStreamsConstants.DATASTORE_PARAM_NAME);
+ if (isNullOrEmpty(datastoreParam)) {
+ final String message = "Stream name does not contain datastore value (pattern /datastore=)";
LOG.debug(message);
throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
}
- final DataChangeScope scope = parseURIEnum(
- DataChangeScope.class,
- mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME));
- if (scope == null) {
- final String message = "Stream name doesn't contains datastore value (pattern /scope=)";
+ // FIXME: this is kept only for compatibility, we are not using this parameter
+ if (isNullOrEmpty(mapOfValues.get(RestconfStreamsConstants.SCOPE_PARAM_NAME))) {
+ final String message = "Stream name does not contain scope value (pattern /scope=)";
LOG.warn(message);
throw new RestconfDocumentedException(message, ErrorType.APPLICATION, ErrorTag.MISSING_ATTRIBUTE);
}
final String streamName = ListenersBroker.createStreamNameFromUri(identifier);
- final Optional<ListenerAdapter> listener = ListenersBroker.getInstance().getDataChangeListenerFor(streamName);
- Preconditions.checkArgument(listener.isPresent(), "Listener doesn't exist : " + streamName);
- listener.get().setQueryParams(
+ final ListenerAdapter listener = ListenersBroker.getInstance()
+ .getDataChangeListenerFor(streamName)
+ .orElseThrow(() -> new IllegalArgumentException("Listener does not exist : " + streamName));
+
+ listener.setQueryParams(
notificationQueryParams.getStart(),
notificationQueryParams.getStop().orElse(null),
notificationQueryParams.getFilter().orElse(null),
- false);
- listener.get().setCloseVars(handlersHolder.getTransactionChainHandler(), handlersHolder.getSchemaHandler());
- registration(datastoreType, listener.get(), handlersHolder.getDomDataBrokerHandler().get());
+ false, notificationQueryParams.isSkipNotificationData());
- final URI uri = prepareUriByStreamName(uriInfo, streamName);
- final DOMTransactionChain transactionChain = handlersHolder.getTransactionChainHandler().get();
- final DOMDataTreeReadWriteTransaction writeTransaction = transactionChain.newReadWriteTransaction();
- final EffectiveModelContext schemaContext = handlersHolder.getSchemaHandler().get();
- final boolean exist = checkExist(schemaContext, writeTransaction);
+ final DOMDataBroker dataBroker = handlersHolder.getDataBroker();
+ final SchemaContextHandler schemaHandler = handlersHolder.getSchemaHandler();
+ listener.setCloseVars(dataBroker, schemaHandler);
+ listener.listen(dataBroker, LogicalDatastoreType.valueOf(datastoreParam));
- final NormalizedNode<?, ?> mapToStreams = RestconfMappingNodeUtil
- .mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.get().getPath(),
- notificationQueryParams.getStart(), listener.get().getOutputType(), uri,
- getMonitoringModule(schemaContext), exist, schemaContext);
- writeDataToDS(schemaContext, listener.get().getPath().getLastPathArgument().getNodeType().getLocalName(),
- writeTransaction, exist, mapToStreams);
+ final URI uri = prepareUriByStreamName(uriInfo, streamName);
+ final EffectiveModelContext schemaContext = schemaHandler.get();
+ final String serializedPath = IdentifierCodec.serialize(listener.getPath(), schemaContext);
+
+ final MapEntryNode mapToStreams =
+ RestconfMappingNodeUtil.mapDataChangeNotificationStreamByIetfRestconfMonitoring(listener.getPath(),
+ notificationQueryParams.getStart(), listener.getOutputType(), uri, schemaContext, serializedPath);
+ final DOMDataTreeWriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
+ writeDataToDS(writeTransaction, mapToStreams);
submitData(writeTransaction);
- transactionChain.close();
return uri;
}
- static Module getMonitoringModule(final SchemaContext schemaContext) {
- return schemaContext.findModule(MonitoringModule.MODULE_QNAME).orElse(null);
+ // FIXME: callers are utter duplicates, refactor them
+ private static void writeDataToDS(final DOMDataTreeWriteOperations tx, final MapEntryNode mapToStreams) {
+ // FIXME: use put() here
+ tx.merge(LogicalDatastoreType.OPERATIONAL, Rfc8040.restconfStateStreamPath(mapToStreams.getIdentifier()),
+ mapToStreams);
}
- /**
- * Parse input of query parameters - start-time or stop-time - from {@link DateAndTime} format
- * to {@link Instant} format.
- *
- * @param entry Start-time or stop-time as string in {@link DateAndTime} format.
- * @return Parsed {@link Instant} by entry.
- */
- static Instant parseDateFromQueryParam(final Entry<String, List<String>> entry) {
- final DateAndTime event = new DateAndTime(entry.getValue().iterator().next());
- final String value = event.getValue();
- final TemporalAccessor accessor;
- try {
- accessor = FORMATTER.parse(value);
- } catch (final DateTimeParseException e) {
- throw new RestconfDocumentedException("Cannot parse of value in date: " + value, e);
- }
- return Instant.from(accessor);
- }
-
- static void writeDataToDS(final SchemaContext schemaContext, final String name,
- final DOMDataTreeReadWriteTransaction readWriteTransaction, final boolean exist,
- final NormalizedNode<?, ?> mapToStreams) {
- String pathId;
- if (exist) {
- pathId = MonitoringModule.PATH_TO_STREAM_WITHOUT_KEY + name;
- } else {
- pathId = MonitoringModule.PATH_TO_STREAMS;
- }
- readWriteTransaction.merge(LogicalDatastoreType.OPERATIONAL,
- IdentifierCodec.deserialize(pathId, schemaContext), mapToStreams);
- }
-
- static void submitData(final DOMDataTreeReadWriteTransaction readWriteTransaction) {
+ private static void submitData(final DOMDataTreeWriteTransaction readWriteTransaction) {
try {
readWriteTransaction.commit().get();
} catch (final InterruptedException | ExecutionException e) {
private static Map<String, String> mapValuesFromUri(final String identifier) {
final HashMap<String, String> result = new HashMap<>();
for (final String token : RestconfConstants.SLASH_SPLITTER.split(identifier)) {
- final String[] paramToken = token.split(String.valueOf(RestconfStreamsConstants.EQUAL));
+ final String[] paramToken = token.split("=");
if (paramToken.length == 2) {
result.put(paramToken[0], paramToken[1]);
}
}
return result;
}
-
- static URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
- final String scheme = uriInfo.getAbsolutePath().getScheme();
- final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
- switch (scheme) {
- case "https":
- // Secured HTTP goes to Secured WebSockets
- uriBuilder.scheme("wss");
- break;
- case "http":
- default:
- // Unsecured HTTP and others go to unsecured WebSockets
- uriBuilder.scheme("ws");
- }
- return uriBuilder.replacePath(RestconfConstants.BASE_URI_PATTERN + RestconfConstants.SLASH + streamName)
- .build();
- }
-
- /**
- * Register data change listener in DOM data broker and set it to listener on stream.
- *
- * @param datastore {@link LogicalDatastoreType}
- * @param listener listener on specific stream
- * @param domDataBroker data broker for register data change listener
- */
- private static void registration(final LogicalDatastoreType datastore, final ListenerAdapter listener,
- final DOMDataBroker domDataBroker) {
- if (listener.isListening()) {
- return;
- }
-
- final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
- .getInstance(DOMDataTreeChangeService.class);
- if (changeService == null) {
- throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
- }
-
- final DOMDataTreeIdentifier root = new DOMDataTreeIdentifier(datastore, listener.getPath());
- final ListenerRegistration<ListenerAdapter> registration =
- changeService.registerDataTreeChangeListener(root, listener);
- listener.setRegistration(registration);
- }
-
- static boolean checkExist(final SchemaContext schemaContext,
- final DOMDataTreeReadOperations readWriteTransaction) {
- boolean exist;
- try {
- return readWriteTransaction.exists(LogicalDatastoreType.OPERATIONAL,
- IdentifierCodec.deserialize(MonitoringModule.PATH_TO_STREAMS, schemaContext)).get();
- } catch (final InterruptedException | ExecutionException exception) {
- throw new RestconfDocumentedException("Problem while checking data if exists", exception);
- }
- }
-
- private static void registerToListenNotification(final NotificationListenerAdapter listener,
- final NotificationServiceHandler notificationServiceHandler) {
- if (listener.isListening()) {
- return;
- }
-
- final SchemaPath path = listener.getSchemaPath();
- final ListenerRegistration<DOMNotificationListener> registration =
- notificationServiceHandler.get().registerNotificationListener(listener, path);
- listener.setRegistration(registration);
- }
-
- /**
- * Parse out enumeration from URI.
- *
- * @param clazz Target enumeration type.
- * @param value String representation of enumeration value.
- * @return Parsed enumeration type.
- */
- private static <T> T parseURIEnum(final Class<T> clazz, final String value) {
- if (value == null || value.equals("")) {
- return null;
- }
- return ResolveEnumUtil.resolveEnum(clazz, value);
- }
}
Code Review / netconf.git / blobdiff