*/
package org.opendaylight.restconf.nb.rfc8040.streams.listeners;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import java.io.IOException;
-import java.io.StringWriter;
-import java.io.Writer;
import java.time.Instant;
-import java.util.Collection;
-import javax.xml.stream.XMLStreamException;
-import javax.xml.transform.dom.DOMResult;
-import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
-import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
-import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
-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.DataContainerChild;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter;
-import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWriter;
+import java.util.Optional;
+import org.opendaylight.mdsal.dom.api.DOMNotification;
+import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
+import org.opendaylight.mdsal.dom.api.DOMNotificationService;
+import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
import org.opendaylight.yangtools.yang.data.codec.gson.JSONCodecFactorySupplier;
-import org.opendaylight.yangtools.yang.data.codec.gson.JSONNormalizedNodeStreamWriter;
-import org.opendaylight.yangtools.yang.data.codec.gson.JsonWriterFactory;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.w3c.dom.Document;
-import org.w3c.dom.Element;
-import org.w3c.dom.Node;
/**
- * {@link NotificationListenerAdapter} is responsible to track events on
- * notifications.
- *
+ * {@link NotificationListenerAdapter} is responsible to track events on notifications.
*/
-public class NotificationListenerAdapter extends AbstractCommonSubscriber implements DOMNotificationListener {
+public final class NotificationListenerAdapter extends AbstractCommonSubscriber<Absolute, DOMNotification>
+ implements DOMNotificationListener {
private static final Logger LOG = LoggerFactory.getLogger(NotificationListenerAdapter.class);
-
- private final String streamName;
- private final SchemaPath path;
- private final String outputType;
-
- private SchemaContext schemaContext;
- private DOMNotification notification;
+ private static final NotificationFormatterFactory JSON_FORMATTER_FACTORY = JSONNotificationFormatter.createFactory(
+ JSONCodecFactorySupplier.RFC7951);
/**
- * Set path of listener and stream name, register event bus.
+ * Set path of listener and stream name.
*
- * @param path
- * path of notification
- * @param streamName
- * stream name of listener
- * @param outputType
- * type of output on notification (JSON, XML)
+ * @param path Schema path of YANG notification.
+ * @param streamName Name of the stream.
+ * @param outputType Type of output on notification (JSON or XML).
*/
- NotificationListenerAdapter(final SchemaPath path, final String streamName, final String outputType) {
- register(this);
- setLocalNameOfPath(path.getLastComponent().getLocalName());
-
- this.outputType = Preconditions.checkNotNull(outputType);
- this.path = Preconditions.checkNotNull(path);
- Preconditions.checkArgument(streamName != null && !streamName.isEmpty());
- this.streamName = streamName;
+ NotificationListenerAdapter(final Absolute path, final String streamName, final NotificationOutputType outputType) {
+ super(path.lastNodeIdentifier(), streamName, path, outputType, getFormatterFactory(outputType));
}
- /**
- * Get outputType of listener.
- *
- * @return the outputType
- */
- @Override
- public String getOutputType() {
- return this.outputType;
+ private static NotificationFormatterFactory getFormatterFactory(final NotificationOutputType outputType) {
+ switch (outputType) {
+ case JSON:
+ return JSON_FORMATTER_FACTORY;
+ case XML:
+ return XMLNotificationFormatter.FACTORY;
+ default:
+ throw new IllegalArgumentException("Unsupported outputType " + outputType);
+ }
}
@Override
- @SuppressWarnings("checkstyle:hiddenField")
+ @SuppressWarnings("checkstyle:IllegalCatch")
public void onNotification(final DOMNotification notification) {
- this.schemaContext = schemaHandler.get();
- this.notification = notification;
-
- final String xml = prepareXml();
- if (checkQueryParams(xml, this)) {
- prepareAndPostData(xml);
+ final Instant now = Instant.now();
+ if (!checkStartStop(now)) {
+ return;
}
- }
-
- /**
- * Get stream name of this listener.
- *
- * @return {@link String}
- */
- @Override
- public String getStreamName() {
- return this.streamName;
- }
-
- /**
- * Get schema path of notification.
- *
- * @return {@link SchemaPath}
- */
- public SchemaPath getSchemaPath() {
- return this.path;
- }
- /**
- * Prepare data of notification and data to client.
- *
- * @param xml data
- */
- private void prepareAndPostData(final String xml) {
- final Event event = new Event(EventType.NOTIFY);
- if (this.outputType.equals("JSON")) {
- event.setData(prepareJson());
- } else {
- event.setData(xml);
+ final Optional<String> maybeOutput;
+ try {
+ maybeOutput = formatter().eventData(schemaHandler.get(), notification, now, getLeafNodesOnly(),
+ isSkipNotificationData());
+ } catch (Exception e) {
+ LOG.error("Failed to process notification {}", notification, e);
+ return;
+ }
+ if (maybeOutput.isPresent()) {
+ post(maybeOutput.get());
}
- post(event);
}
/**
- * Prepare json from notification data.
+ * Get schema path of notification.
*
- * @return json as {@link String}
+ * @return The configured schema path that points to observing YANG notification schema node.
*/
- @VisibleForTesting
- String prepareJson() {
- final JsonParser jsonParser = new JsonParser();
- final JsonObject json = new JsonObject();
- json.add("ietf-restconf:notification", jsonParser.parse(writeBodyToString()));
- json.addProperty("event-time", ListenerAdapter.toRFC3339(Instant.now()));
- return json.toString();
- }
-
- @VisibleForTesting
- void setNotification(final DOMNotification notification) {
- this.notification = Preconditions.checkNotNull(notification);
- }
-
- @VisibleForTesting
- void setSchemaContext(final SchemaContext schemaContext) {
- this.schemaContext = Preconditions.checkNotNull(schemaContext);
- }
-
- private String writeBodyToString() {
- final Writer writer = new StringWriter();
- final NormalizedNodeStreamWriter jsonStream = JSONNormalizedNodeStreamWriter.createExclusiveWriter(
- JSONCodecFactorySupplier.DRAFT_LHOTKA_NETMOD_YANG_JSON_02.getShared(this.schemaContext),
- this.notification.getType(), null, JsonWriterFactory.createJsonWriter(writer));
- final NormalizedNodeWriter nodeWriter = NormalizedNodeWriter.forStreamWriter(jsonStream);
- try {
- nodeWriter.write(this.notification.getBody());
- nodeWriter.close();
- } catch (final IOException e) {
- throw new RestconfDocumentedException("Problem while writing body of notification to JSON. ", e);
- }
- return writer.toString();
- }
-
- private String prepareXml() {
- final Document doc = createDocument();
- final Element notificationElement = basePartDoc(doc);
-
- final Element notificationEventElement = doc.createElementNS(
- "urn:opendaylight:params:xml:ns:yang:controller:md:sal:remote", "create-notification-stream");
- addValuesToNotificationEventElement(doc, notificationEventElement);
- notificationElement.appendChild(notificationEventElement);
-
- return transformDoc(doc);
+ public Absolute getSchemaPath() {
+ return path();
}
- private void addValuesToNotificationEventElement(final Document doc, final Element element) {
- if (notification == null) {
- return;
- }
-
- final NormalizedNode<NodeIdentifier, Collection<DataContainerChild<? extends PathArgument, ?>>> body =
- notification.getBody();
- try {
-
- final DOMResult domResult = writeNormalizedNode(body, schemaContext, this.path);
- final Node result = doc.importNode(domResult.getNode().getFirstChild(), true);
- final Element dataElement = doc.createElement("notification");
- dataElement.appendChild(result);
- element.appendChild(dataElement);
- } catch (final IOException e) {
- LOG.error("Error in writer ", e);
- } catch (final XMLStreamException e) {
- LOG.error("Error processing stream", e);
+ public synchronized void listen(final DOMNotificationService notificationService) {
+ if (!isListening()) {
+ setRegistration(notificationService.registerNotificationListener(this, getSchemaPath()));
}
}
}
Code Review / netconf.git / blobdiff