X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-rest-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fstreams%2Flisteners%2FListenerAdapter.java;h=bd33a2f40fe982431159cd6e79b68d1c1c53ef1e;hb=e631dc96f0461b2270377dc072b9f969a875667a;hp=925a09337ca3ee1e8fba707020e3c7a19b8e03f6;hpb=0e7b83c0739ac579ca6dbf6a3f7992f70eb3000f;p=controller.git diff --git a/opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/listeners/ListenerAdapter.java b/opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/listeners/ListenerAdapter.java index 925a09337c..bd33a2f40f 100644 --- a/opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/listeners/ListenerAdapter.java +++ b/opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/listeners/ListenerAdapter.java @@ -12,11 +12,9 @@ import com.google.common.base.Preconditions; import com.google.common.eventbus.AsyncEventBus; import com.google.common.eventbus.EventBus; import com.google.common.eventbus.Subscribe; - import io.netty.channel.Channel; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.util.internal.ConcurrentSet; - import java.io.ByteArrayOutputStream; import java.io.OutputStreamWriter; import java.io.UnsupportedEncodingException; @@ -25,13 +23,10 @@ import java.util.Collection; import java.util.Date; import java.util.HashMap; import java.util.Map; -import java.util.Map.Entry; import java.util.Random; import java.util.Set; import java.util.concurrent.Executors; import java.util.regex.Pattern; - -import javax.activation.UnsupportedDataTypeException; import javax.xml.parsers.DocumentBuilder; import javax.xml.parsers.DocumentBuilderFactory; import javax.xml.parsers.ParserConfigurationException; @@ -41,19 +36,16 @@ import javax.xml.transform.TransformerException; import javax.xml.transform.TransformerFactory; import javax.xml.transform.dom.DOMSource; import javax.xml.transform.stream.StreamResult; - -import org.opendaylight.controller.md.sal.common.api.data.DataChangeEvent; -import org.opendaylight.controller.sal.core.api.data.DataChangeListener; -import org.opendaylight.controller.sal.rest.impl.XmlMapper; +import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent; +import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener; import org.opendaylight.controller.sal.restconf.impl.ControllerContext; import org.opendaylight.yangtools.concepts.ListenerRegistration; import org.opendaylight.yangtools.yang.common.QName; -import org.opendaylight.yangtools.yang.data.api.CompositeNode; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.NodeIdentifierWithPredicates; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.NodeWithValue; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.PathArgument; -import org.opendaylight.yangtools.yang.model.api.DataNodeContainer; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeWithValue; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument; +import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.w3c.dom.Document; @@ -61,41 +53,35 @@ import org.w3c.dom.Element; import org.w3c.dom.Node; /** - * {@link ListenerAdapter} is responsible to track events, which occurred by - * changing data in data source. + * {@link ListenerAdapter} is responsible to track events, which occurred by changing data in data source. */ -public class ListenerAdapter implements DataChangeListener { +public class ListenerAdapter implements DOMDataChangeListener { private static final Logger LOG = LoggerFactory.getLogger(ListenerAdapter.class); private static final DocumentBuilderFactory DBF = DocumentBuilderFactory.newInstance(); private static final TransformerFactory FACTORY = TransformerFactory.newInstance(); private static final Pattern RFC3339_PATTERN = Pattern.compile("(\\d\\d)(\\d\\d)$"); - private final XmlMapper xmlMapper = new XmlMapper(); - private final SimpleDateFormat rfc3339 = new SimpleDateFormat( - "yyyy-MM-dd'T'hh:mm:ssZ"); + private final SimpleDateFormat rfc3339 = new SimpleDateFormat("yyyy-MM-dd'T'hh:mm:ssZ"); - private final InstanceIdentifier path; - private ListenerRegistration registration; + private final YangInstanceIdentifier path; + private ListenerRegistration registration; private final String streamName; private Set subscribers = new ConcurrentSet<>(); private final EventBus eventBus; private final EventBusChangeRecorder eventBusChangeRecorder; - /** - * Creates new {@link ListenerAdapter} listener specified by path and stream - * name. + * Creates new {@link ListenerAdapter} listener specified by path and stream name. * * @param path * Path to data in data store. * @param streamName * The name of the stream. */ - ListenerAdapter(final InstanceIdentifier path, final String streamName) { + ListenerAdapter(final YangInstanceIdentifier path, final String streamName) { Preconditions.checkNotNull(path); - Preconditions - .checkArgument(streamName != null && !streamName.isEmpty()); + Preconditions.checkArgument(streamName != null && !streamName.isEmpty()); this.path = path; this.streamName = streamName; eventBus = new AsyncEventBus(Executors.newSingleThreadExecutor()); @@ -104,16 +90,13 @@ public class ListenerAdapter implements DataChangeListener { } @Override - public void onDataChanged( - final DataChangeEvent change) { - if (!change.getCreatedConfigurationData().isEmpty() - || !change.getCreatedOperationalData().isEmpty() - || !change.getUpdatedConfigurationData().isEmpty() - || !change.getUpdatedOperationalData().isEmpty() - || !change.getRemovedConfigurationData().isEmpty() - || !change.getRemovedOperationalData().isEmpty()) { - String xml = prepareXmlFrom(change); - Event event = new Event(EventType.NOTIFY); + public void onDataChanged(final AsyncDataChangeEvent> change) { + // TODO Auto-generated method stub + + if (!change.getCreatedData().isEmpty() || !change.getUpdatedData().isEmpty() + || !change.getRemovedPaths().isEmpty()) { + final String xml = prepareXmlFrom(change); + final Event event = new Event(EventType.NOTIFY); event.setData(xml); eventBus.post(event); } @@ -126,25 +109,20 @@ public class ListenerAdapter implements DataChangeListener { @Subscribe public void recordCustomerChange(final Event event) { if (event.getType() == EventType.REGISTER) { - Channel subscriber = event.getSubscriber(); + final Channel subscriber = event.getSubscriber(); if (!subscribers.contains(subscriber)) { subscribers.add(subscriber); } } else if (event.getType() == EventType.DEREGISTER) { subscribers.remove(event.getSubscriber()); - Notificator - .removeListenerIfNoSubscriberExists(ListenerAdapter.this); + Notificator.removeListenerIfNoSubscriberExists(ListenerAdapter.this); } else if (event.getType() == EventType.NOTIFY) { - for (Channel subscriber : subscribers) { + for (final Channel subscriber : subscribers) { if (subscriber.isActive()) { - LOG.debug("Data are sent to subscriber {}:", - subscriber.remoteAddress()); - subscriber.writeAndFlush(new TextWebSocketFrame(event - .getData())); + LOG.debug("Data are sent to subscriber {}:", subscriber.remoteAddress()); + subscriber.writeAndFlush(new TextWebSocketFrame(event.getData())); } else { - LOG.debug( - "Subscriber {} is removed - channel is not active yet.", - subscriber.remoteAddress()); + LOG.debug("Subscriber {} is removed - channel is not active yet.", subscriber.remoteAddress()); subscribers.remove(subscriber); } } @@ -153,8 +131,7 @@ public class ListenerAdapter implements DataChangeListener { } /** - * Represents event of specific {@link EventType} type, holds data and - * {@link Channel} subscriber. + * Represents event of specific {@link EventType} type, holds data and {@link Channel} subscriber. */ private final class Event { private final EventType type; @@ -191,7 +168,7 @@ public class ListenerAdapter implements DataChangeListener { } /** - * Gets event data. + * Gets event String. * * @return String representation of event data. */ @@ -202,8 +179,7 @@ public class ListenerAdapter implements DataChangeListener { /** * Sets event data. * - * @param String - * data. + * @param data String. */ public void setData(final String data) { this.data = data; @@ -223,7 +199,9 @@ public class ListenerAdapter implements DataChangeListener { * Type of the event. */ private enum EventType { - REGISTER, DEREGISTER, NOTIFY; + REGISTER, + DEREGISTER, + NOTIFY; } /** @@ -233,39 +211,34 @@ public class ListenerAdapter implements DataChangeListener { * DataChangeEvent * @return Data in printable form. */ - private String prepareXmlFrom( - final DataChangeEvent change) { - Document doc = createDocument(); - Element notificationElement = doc.createElementNS( - "urn:ietf:params:xml:ns:netconf:notification:1.0", + private String prepareXmlFrom(final AsyncDataChangeEvent> change) { + final Document doc = createDocument(); + final Element notificationElement = doc.createElementNS("urn:ietf:params:xml:ns:netconf:notification:1.0", "notification"); doc.appendChild(notificationElement); - Element eventTimeElement = doc.createElement("eventTime"); + final Element eventTimeElement = doc.createElement("eventTime"); eventTimeElement.setTextContent(toRFC3339(new Date())); notificationElement.appendChild(eventTimeElement); - Element dataChangedNotificationEventElement = doc.createElementNS( - "urn:opendaylight:params:xml:ns:yang:controller:md:sal:remote", - "data-changed-notification"); - addValuesToDataChangedNotificationEventElement(doc, - dataChangedNotificationEventElement, change); + final Element dataChangedNotificationEventElement = doc.createElementNS( + "urn:opendaylight:params:xml:ns:yang:controller:md:sal:remote", "data-changed-notification"); + addValuesToDataChangedNotificationEventElement(doc, dataChangedNotificationEventElement, change); notificationElement.appendChild(dataChangedNotificationEventElement); try { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - Transformer transformer = FACTORY.newTransformer(); + final ByteArrayOutputStream out = new ByteArrayOutputStream(); + final Transformer transformer = FACTORY.newTransformer(); transformer.setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "no"); transformer.setOutputProperty(OutputKeys.METHOD, "xml"); transformer.setOutputProperty(OutputKeys.INDENT, "yes"); transformer.setOutputProperty(OutputKeys.ENCODING, "UTF-8"); transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount", "4"); - transformer.transform(new DOMSource(doc), new StreamResult( - new OutputStreamWriter(out, Charsets.UTF_8))); - byte[] charData = out.toByteArray(); + transformer.transform(new DOMSource(doc), new StreamResult(new OutputStreamWriter(out, Charsets.UTF_8))); + final byte[] charData = out.toByteArray(); return new String(charData, "UTF-8"); } catch (TransformerException | UnsupportedEncodingException e) { - String msg = "Error during transformation of Document into String"; + final String msg = "Error during transformation of Document into String"; LOG.error(msg, e); return msg; } @@ -284,14 +257,13 @@ public class ListenerAdapter implements DataChangeListener { /** * Creates {@link Document} document. - * * @return {@link Document} document. */ private Document createDocument() { final DocumentBuilder bob; try { bob = DBF.newDocumentBuilder(); - } catch (ParserConfigurationException e) { + } catch (final ParserConfigurationException e) { return null; } return bob.newDocument(); @@ -305,33 +277,18 @@ public class ListenerAdapter implements DataChangeListener { * @param dataChangedNotificationEventElement * {@link Element} * @param change - * {@link DataChangeEvent} + * {@link AsyncDataChangeEvent} */ private void addValuesToDataChangedNotificationEventElement(final Document doc, final Element dataChangedNotificationEventElement, - final DataChangeEvent change) { - addValuesFromDataToElement(doc, change.getCreatedConfigurationData(), - dataChangedNotificationEventElement, Store.CONFIG, + final AsyncDataChangeEvent> change) { + addValuesFromDataToElement(doc, change.getCreatedData().keySet(), dataChangedNotificationEventElement, Operation.CREATED); - addValuesFromDataToElement(doc, change.getCreatedOperationalData(), - dataChangedNotificationEventElement, Store.OPERATION, - Operation.CREATED); - if (change.getCreatedConfigurationData().isEmpty()) { - addValuesFromDataToElement(doc, - change.getUpdatedConfigurationData(), - dataChangedNotificationEventElement, Store.CONFIG, - Operation.UPDATED); - } - if (change.getCreatedOperationalData().isEmpty()) { - addValuesFromDataToElement(doc, change.getUpdatedOperationalData(), - dataChangedNotificationEventElement, Store.OPERATION, + if (change.getCreatedData().isEmpty()) { + addValuesFromDataToElement(doc, change.getUpdatedData().keySet(), dataChangedNotificationEventElement, Operation.UPDATED); } - addValuesFromDataToElement(doc, change.getRemovedConfigurationData(), - dataChangedNotificationEventElement, Store.CONFIG, - Operation.DELETED); - addValuesFromDataToElement(doc, change.getRemovedOperationalData(), - dataChangedNotificationEventElement, Store.OPERATION, + addValuesFromDataToElement(doc, change.getRemovedPaths(), dataChangedNotificationEventElement, Operation.DELETED); } @@ -341,53 +298,25 @@ public class ListenerAdapter implements DataChangeListener { * @param doc * {@link Document} * @param data - * Set of {@link InstanceIdentifier}. + * Set of {@link YangInstanceIdentifier}. * @param element * {@link Element} - * @param store - * {@link Store} * @param operation * {@link Operation} */ - private void addValuesFromDataToElement(final Document doc, - final Set data, final Element element, final Store store, + private void addValuesFromDataToElement(final Document doc, final Set data, final Element element, final Operation operation) { if (data == null || data.isEmpty()) { return; } - for (InstanceIdentifier path : data) { - Node node = createDataChangeEventElement(doc, path, null, store, - operation); - element.appendChild(node); + for (final YangInstanceIdentifier path : data) { + if (!ControllerContext.getInstance().isNodeMixin(path)) { + final Node node = createDataChangeEventElement(doc, path, operation); + element.appendChild(node); + } } } - /** - * Adds values from data to element. - * - * @param doc - * {@link Document} - * @param data - * Map of {@link InstanceIdentifier} and {@link CompositeNode}. - * @param element - * {@link Element} - * @param store - * {@link Store} - * @param operation - * {@link Operation} - */ - private void addValuesFromDataToElement(final Document doc, - final Map data, final Element element, - final Store store, final Operation operation) { - if (data == null || data.isEmpty()) { - return; - } - for (Entry entry : data.entrySet()) { - Node node = createDataChangeEventElement(doc, entry.getKey(), - entry.getValue(), store, operation); - element.appendChild(node); - } - } /** * Creates changed event element from data. @@ -396,70 +325,23 @@ public class ListenerAdapter implements DataChangeListener { * {@link Document} * @param path * Path to data in data store. - * @param data - * {@link CompositeNode} - * @param store - * {@link Store} * @param operation * {@link Operation} * @return {@link Node} node represented by changed event element. */ - private Node createDataChangeEventElement(final Document doc, - final InstanceIdentifier path, final CompositeNode data, final Store store, - final Operation operation) { - Element dataChangeEventElement = doc.createElement("data-change-event"); - - Element pathElement = doc.createElement("path"); + private Node createDataChangeEventElement(final Document doc, final YangInstanceIdentifier path, final Operation operation) { + final Element dataChangeEventElement = doc.createElement("data-change-event"); + final Element pathElement = doc.createElement("path"); addPathAsValueToElement(path, pathElement); dataChangeEventElement.appendChild(pathElement); - Element storeElement = doc.createElement("store"); - storeElement.setTextContent(store.value); - dataChangeEventElement.appendChild(storeElement); - - Element operationElement = doc.createElement("operation"); + final Element operationElement = doc.createElement("operation"); operationElement.setTextContent(operation.value); dataChangeEventElement.appendChild(operationElement); - if (data != null) { - Element dataElement = doc.createElement("data"); - Node dataAnyXml = translateToXml(path, data); - Node adoptedNode = doc.adoptNode(dataAnyXml); - dataElement.appendChild(adoptedNode); - dataChangeEventElement.appendChild(dataElement); - } - return dataChangeEventElement; } - /** - * Translates {@link CompositeNode} data to XML format. - * - * @param path - * Path to data in data store. - * @param data - * {@link CompositeNode} - * @return Data in XML format. - */ - private Node translateToXml(final InstanceIdentifier path, final CompositeNode data) { - DataNodeContainer schemaNode = ControllerContext.getInstance() - .getDataNodeContainerFor(path); - if (schemaNode == null) { - LOG.info( - "Path '{}' contains node with unsupported type (supported type is Container or List) or some node was not found.", - path); - return null; - } - try { - Document xml = xmlMapper.write(data, schemaNode); - return xml.getFirstChild(); - } catch (UnsupportedDataTypeException e) { - LOG.error( - "Error occured during translation of notification to XML.", - e); - return null; - } - } /** * Adds path as value to element. @@ -469,27 +351,25 @@ public class ListenerAdapter implements DataChangeListener { * @param element * {@link Element} */ - private void addPathAsValueToElement(final InstanceIdentifier path, - final Element element) { + private void addPathAsValueToElement(final YangInstanceIdentifier path, final Element element) { // Map< key = namespace, value = prefix> - Map prefixes = new HashMap<>(); - InstanceIdentifier instanceIdentifier = path; - StringBuilder textContent = new StringBuilder(); + final Map prefixes = new HashMap<>(); + final YangInstanceIdentifier normalizedPath = ControllerContext.getInstance().toXpathRepresentation(path); + final StringBuilder textContent = new StringBuilder(); // FIXME: BUG-1281: this is duplicated code from yangtools (BUG-1275) - for (PathArgument pathArgument : instanceIdentifier.getPathArguments()) { + for (final PathArgument pathArgument : normalizedPath.getPathArguments()) { + if (pathArgument instanceof YangInstanceIdentifier.AugmentationIdentifier) { + continue; + } textContent.append("/"); - writeIdentifierWithNamespacePrefix(element, textContent, - pathArgument.getNodeType(), prefixes); + writeIdentifierWithNamespacePrefix(element, textContent, pathArgument.getNodeType(), prefixes); if (pathArgument instanceof NodeIdentifierWithPredicates) { - Map predicates = ((NodeIdentifierWithPredicates) pathArgument) - .getKeyValues(); - for (QName keyValue : predicates.keySet()) { - String predicateValue = String.valueOf(predicates - .get(keyValue)); + final Map predicates = ((NodeIdentifierWithPredicates) pathArgument).getKeyValues(); + for (final QName keyValue : predicates.keySet()) { + final String predicateValue = String.valueOf(predicates.get(keyValue)); textContent.append("["); - writeIdentifierWithNamespacePrefix(element, textContent, - keyValue, prefixes); + writeIdentifierWithNamespacePrefix(element, textContent, keyValue, prefixes); textContent.append("='"); textContent.append(predicateValue); textContent.append("'"); @@ -517,16 +397,12 @@ public class ListenerAdapter implements DataChangeListener { * @param prefixes * Map of namespaces and prefixes. */ - private static void writeIdentifierWithNamespacePrefix(final Element element, - final StringBuilder textContent, final QName qName, final Map prefixes) { - String namespace = qName.getNamespace().toString(); + private static void writeIdentifierWithNamespacePrefix(final Element element, final StringBuilder textContent, + final QName qName, final Map prefixes) { + final String namespace = qName.getNamespace().toString(); String prefix = prefixes.get(namespace); if (prefix == null) { - prefix = qName.getPrefix(); - if (prefix == null || prefix.isEmpty() - || prefixes.containsValue(prefix)) { - prefix = generateNewPrefix(prefixes.values()); - } + prefix = generateNewPrefix(prefixes.values()); } element.setAttribute("xmlns:" + prefix, namespace); @@ -546,11 +422,11 @@ public class ListenerAdapter implements DataChangeListener { */ private static String generateNewPrefix(final Collection prefixes) { StringBuilder result = null; - Random random = new Random(); + final Random random = new Random(); do { result = new StringBuilder(); for (int i = 0; i < 4; i++) { - int randomNumber = 0x61 + (Math.abs(random.nextInt()) % 26); + final int randomNumber = 0x61 + (Math.abs(random.nextInt()) % 26); result.append(Character.toChars(randomNumber)); } } while (prefixes.contains(result.toString())); @@ -563,7 +439,7 @@ public class ListenerAdapter implements DataChangeListener { * * @return Path pointed to data in data store. */ - public InstanceIdentifier getPath() { + public YangInstanceIdentifier getPath() { return path; } @@ -573,8 +449,7 @@ public class ListenerAdapter implements DataChangeListener { * @param registration * ListenerRegistration */ - public void setRegistration( - final ListenerRegistration registration) { + public void setRegistration(final ListenerRegistration registration) { this.registration = registration; } @@ -588,8 +463,7 @@ public class ListenerAdapter implements DataChangeListener { } /** - * Removes all subscribers and unregisters event bus change recorder form - * event bus. + * Removes all subscribers and unregisters event bus change recorder form event bus. */ public void close() throws Exception { subscribers = new ConcurrentSet<>(); @@ -608,31 +482,30 @@ public class ListenerAdapter implements DataChangeListener { } /** - * Creates event of type {@link EventType#REGISTER}, set {@link Channel} - * subscriber to the event and post event into event bus. + * Creates event of type {@link EventType#REGISTER}, set {@link Channel} subscriber to the event and post event into + * event bus. * * @param subscriber * Channel */ public void addSubscriber(final Channel subscriber) { if (!subscriber.isActive()) { - LOG.debug("Channel is not active between websocket server and subscriber {}" - + subscriber.remoteAddress()); + LOG.debug("Channel is not active between websocket server and subscriber {}" + subscriber.remoteAddress()); } - Event event = new Event(EventType.REGISTER); + final Event event = new Event(EventType.REGISTER); event.setSubscriber(subscriber); eventBus.post(event); } /** - * Creates event of type {@link EventType#DEREGISTER}, sets {@link Channel} - * subscriber to the event and posts event into event bus. + * Creates event of type {@link EventType#DEREGISTER}, sets {@link Channel} subscriber to the event and posts event + * into event bus. * * @param subscriber */ public void removeSubscriber(final Channel subscriber) { LOG.debug("Subscriber {} is removed.", subscriber.remoteAddress()); - Event event = new Event(EventType.DEREGISTER); + final Event event = new Event(EventType.DEREGISTER); event.setSubscriber(subscriber); eventBus.post(event); } @@ -640,8 +513,7 @@ public class ListenerAdapter implements DataChangeListener { /** * Checks if exists at least one {@link Channel} subscriber. * - * @return True if exist at least one {@link Channel} subscriber, false - * otherwise. + * @return True if exist at least one {@link Channel} subscriber, false otherwise. */ public boolean hasSubscribers() { return !subscribers.isEmpty(); @@ -651,7 +523,8 @@ public class ListenerAdapter implements DataChangeListener { * Consists of two types {@link Store#CONFIG} and {@link Store#OPERATION}. */ private static enum Store { - CONFIG("config"), OPERATION("operation"); + CONFIG("config"), + OPERATION("operation"); private final String value; @@ -661,11 +534,12 @@ public class ListenerAdapter implements DataChangeListener { } /** - * Consists of three types {@link Operation#CREATED}, - * {@link Operation#UPDATED} and {@link Operation#DELETED}. + * Consists of three types {@link Operation#CREATED}, {@link Operation#UPDATED} and {@link Operation#DELETED}. */ private static enum Operation { - CREATED("created"), UPDATED("updated"), DELETED("deleted"); + CREATED("created"), + UPDATED("updated"), + DELETED("deleted"); private final String value;