restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/listeners/ListenerAdapter.java

   1 /*
   2  * Copyright (c) 2014, 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.restconf.nb.rfc8040.streams.listeners;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.annotations.VisibleForTesting;
  14 import com.google.common.base.MoreObjects;
  15 import java.time.Instant;
  16 import java.util.Collection;
  17 import java.util.Optional;
  18 import java.util.stream.Collectors;
  19 import javax.xml.xpath.XPathExpressionException;
  20 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  21 import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
  22 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  23 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
  24 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  25 import org.opendaylight.restconf.common.formatters.DataTreeCandidateFormatter;
  26 import org.opendaylight.restconf.common.formatters.DataTreeCandidateFormatterFactory;
  27 import org.opendaylight.restconf.common.formatters.JSONDataTreeCandidateFormatter;
  28 import org.opendaylight.restconf.common.formatters.XMLDataTreeCandidateFormatter;
  29 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  30 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  31 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  32 import org.opendaylight.yangtools.yang.data.codec.gson.JSONCodecFactorySupplier;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * {@link ListenerAdapter} is responsible to track events, which occurred by changing data in data source.
  38  */
  39 public class ListenerAdapter extends AbstractCommonSubscriber implements ClusteredDOMDataTreeChangeListener {
  40     private static final Logger LOG = LoggerFactory.getLogger(ListenerAdapter.class);
  41     private static final String PATH = "path";
  42     private static final DataTreeCandidateFormatterFactory JSON_FORMATTER_FACTORY =
  43             JSONDataTreeCandidateFormatter.createFactory(JSONCodecFactorySupplier.RFC7951);
  44
  45     private final YangInstanceIdentifier path;
  46     private final String streamName;
  47     private final NotificationOutputType outputType;
  48
  49     @VisibleForTesting DataTreeCandidateFormatter formatter;
  50
  51     /**
  52      * Creates new {@link ListenerAdapter} listener specified by path and stream name and register for subscribing.
  53      *
  54      * @param path       Path to data in data store.
  55      * @param streamName The name of the stream.
  56      * @param outputType Type of output on notification (JSON, XML).
  57      */
  58     ListenerAdapter(final YangInstanceIdentifier path, final String streamName,
  59             final NotificationOutputType outputType) {
  60         setLocalNameOfPath(path.getLastPathArgument().getNodeType().getLocalName());
  61
  62         this.outputType = requireNonNull(outputType);
  63         this.path = requireNonNull(path);
  64         this.streamName = requireNonNull(streamName);
  65         checkArgument(!streamName.isEmpty());
  66
  67         formatter = getFormatterFactory().getFormatter();
  68     }
  69
  70     private DataTreeCandidateFormatterFactory getFormatterFactory() {
  71         switch (outputType) {
  72             case JSON:
  73                 return JSON_FORMATTER_FACTORY;
  74             case XML:
  75                 return XMLDataTreeCandidateFormatter.FACTORY;
  76             default:
  77                 throw new IllegalArgumentException("Unsupported outputType" + outputType);
  78         }
  79     }
  80
  81     private DataTreeCandidateFormatter getFormatter(final String filter) throws XPathExpressionException {
  82         final DataTreeCandidateFormatterFactory factory = getFormatterFactory();
  83         return filter == null || filter.isEmpty() ? factory.getFormatter() : factory.getFormatter(filter);
  84     }
  85
  86     @Override
  87     public void setQueryParams(final Instant startTime, final Instant stopTime, final String filter,
  88                                final boolean leafNodesOnly, final boolean skipNotificationData) {
  89         setQueryParams(startTime, stopTime, leafNodesOnly, skipNotificationData);
  90         try {
  91             formatter = getFormatter(filter);
  92         } catch (final XPathExpressionException e) {
  93             throw new IllegalArgumentException("Failed to get filter", e);
  94         }
  95     }
  96
  97     @Override
  98     public void onInitialData() {
  99         // No-op
 100     }
 101
 102     @Override
 103     @SuppressWarnings("checkstyle:IllegalCatch")
 104     public void onDataTreeChanged(final Collection<DataTreeCandidate> dataTreeCandidates) {
 105         final Instant now = Instant.now();
 106         if (!checkStartStop(now, this)) {
 107             return;
 108         }
 109
 110         final Optional<String> maybeData;
 111         try {
 112             maybeData = formatter.eventData(schemaHandler.get(), dataTreeCandidates, now, getLeafNodesOnly(),
 113                     isSkipNotificationData());
 114         } catch (final Exception e) {
 115             LOG.error("Failed to process notification {}",
 116                     dataTreeCandidates.stream().map(Object::toString).collect(Collectors.joining(",")), e);
 117             return;
 118         }
 119
 120         if (maybeData.isPresent()) {
 121             post(maybeData.get());
 122         }
 123     }
 124
 125     /**
 126      * Gets the name of the stream.
 127      *
 128      * @return The name of the stream.
 129      */
 130     @Override
 131     public String getStreamName() {
 132         return streamName;
 133     }
 134
 135     @Override
 136     public String getOutputType() {
 137         return outputType.getName();
 138     }
 139
 140     /**
 141      * Get path pointed to data in data store.
 142      *
 143      * @return Path pointed to data in data store.
 144      */
 145     public YangInstanceIdentifier getPath() {
 146         return path;
 147     }
 148
 149     /**
 150      * Register data change listener in DOM data broker and set it to listener on stream.
 151      *
 152      * @param domDataBroker data broker for register data change listener
 153      * @param datastore     {@link LogicalDatastoreType}
 154      */
 155     public final synchronized void listen(final DOMDataBroker domDataBroker, final LogicalDatastoreType datastore) {
 156         if (!isListening()) {
 157             final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
 158                 .getInstance(DOMDataTreeChangeService.class);
 159             if (changeService == null) {
 160                 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 161             }
 162
 163             setRegistration(changeService.registerDataTreeChangeListener(
 164                 new DOMDataTreeIdentifier(datastore, getPath()), this));
 165         }
 166     }
 167
 168     @Override
 169     public String toString() {
 170         return MoreObjects.toStringHelper(this)
 171                 .add(PATH, path)
 172                 .add("stream-name", streamName)
 173                 .add("output-type", outputType)
 174                 .toString();
 175     }
 176 }