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     @Override
  82     final void setFilter(final String filter) throws XPathExpressionException {
  83         final DataTreeCandidateFormatterFactory factory = getFormatterFactory();
  84         formatter = filter == null || filter.isEmpty() ? factory.getFormatter() : factory.getFormatter(filter);
  85     }
  86
  87     @Override
  88     public void onInitialData() {
  89         // No-op
  90     }
  91
  92     @Override
  93     @SuppressWarnings("checkstyle:IllegalCatch")
  94     public void onDataTreeChanged(final Collection<DataTreeCandidate> dataTreeCandidates) {
  95         final Instant now = Instant.now();
  96         if (!checkStartStop(now)) {
  97             return;
  98         }
  99
 100         final Optional<String> maybeData;
 101         try {
 102             maybeData = formatter.eventData(schemaHandler.get(), dataTreeCandidates, now, getLeafNodesOnly(),
 103                     isSkipNotificationData());
 104         } catch (final Exception e) {
 105             LOG.error("Failed to process notification {}",
 106                     dataTreeCandidates.stream().map(Object::toString).collect(Collectors.joining(",")), e);
 107             return;
 108         }
 109
 110         if (maybeData.isPresent()) {
 111             post(maybeData.get());
 112         }
 113     }
 114
 115     /**
 116      * Gets the name of the stream.
 117      *
 118      * @return The name of the stream.
 119      */
 120     @Override
 121     public String getStreamName() {
 122         return streamName;
 123     }
 124
 125     @Override
 126     public String getOutputType() {
 127         return outputType.getName();
 128     }
 129
 130     /**
 131      * Get path pointed to data in data store.
 132      *
 133      * @return Path pointed to data in data store.
 134      */
 135     public YangInstanceIdentifier getPath() {
 136         return path;
 137     }
 138
 139     /**
 140      * Register data change listener in DOM data broker and set it to listener on stream.
 141      *
 142      * @param domDataBroker data broker for register data change listener
 143      * @param datastore     {@link LogicalDatastoreType}
 144      */
 145     public final synchronized void listen(final DOMDataBroker domDataBroker, final LogicalDatastoreType datastore) {
 146         if (!isListening()) {
 147             final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
 148                 .getInstance(DOMDataTreeChangeService.class);
 149             if (changeService == null) {
 150                 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
 151             }
 152
 153             setRegistration(changeService.registerDataTreeChangeListener(
 154                 new DOMDataTreeIdentifier(datastore, getPath()), this));
 155         }
 156     }
 157
 158     @Override
 159     public String toString() {
 160         return MoreObjects.toStringHelper(this)
 161                 .add(PATH, path)
 162                 .add("stream-name", streamName)
 163                 .add("output-type", outputType)
 164                 .toString();
 165     }
 166 }