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

   1 /*
   2  * Copyright (c) 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.Optional;
  17 import javax.xml.xpath.XPathExpressionException;
  18 import org.opendaylight.mdsal.dom.api.DOMNotification;
  19 import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
  20 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
  21 import org.opendaylight.restconf.common.formatters.JSONNotificationFormatter;
  22 import org.opendaylight.restconf.common.formatters.NotificationFormatter;
  23 import org.opendaylight.restconf.common.formatters.NotificationFormatterFactory;
  24 import org.opendaylight.restconf.common.formatters.XMLNotificationFormatter;
  25 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  26 import org.opendaylight.yangtools.yang.data.codec.gson.JSONCodecFactorySupplier;
  27 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 /**
  32  * {@link NotificationListenerAdapter} is responsible to track events on notifications.
  33  */
  34 public class NotificationListenerAdapter extends AbstractCommonSubscriber implements DOMNotificationListener {
  35
  36     private static final Logger LOG = LoggerFactory.getLogger(NotificationListenerAdapter.class);
  37     private static final NotificationFormatterFactory JSON_FORMATTER_FACTORY = JSONNotificationFormatter.createFactory(
  38             JSONCodecFactorySupplier.RFC7951);
  39
  40     private final String streamName;
  41     private final Absolute path;
  42     private final NotificationOutputType outputType;
  43
  44     @VisibleForTesting NotificationFormatter formatter;
  45
  46
  47     /**
  48      * Set path of listener and stream name.
  49      *
  50      * @param path       Schema path of YANG notification.
  51      * @param streamName Name of the stream.
  52      * @param outputType Type of output on notification (JSON or XML).
  53      */
  54     NotificationListenerAdapter(final Absolute path, final String streamName, final String outputType) {
  55         setLocalNameOfPath(path.lastNodeIdentifier().getLocalName());
  56
  57         this.outputType = NotificationOutputType.forName(requireNonNull(outputType)).get();
  58         this.path = requireNonNull(path);
  59         this.streamName = requireNonNull(streamName);
  60         checkArgument(!streamName.isEmpty());
  61         this.formatter = getFormatterFactory().getFormatter();
  62
  63         LOG.debug("output type: {}, {}", outputType, this.outputType);
  64     }
  65
  66     private NotificationFormatterFactory getFormatterFactory() {
  67         switch (outputType) {
  68             case JSON:
  69                 return JSON_FORMATTER_FACTORY;
  70             case XML:
  71                 return XMLNotificationFormatter.FACTORY;
  72             default:
  73                 throw new IllegalArgumentException("Unsupported outputType " + outputType);
  74         }
  75     }
  76
  77     private NotificationFormatter getFormatter(final String filter) throws XPathExpressionException {
  78         NotificationFormatterFactory factory = getFormatterFactory();
  79         return filter == null || filter.isEmpty() ? factory.getFormatter() : factory.getFormatter(filter);
  80     }
  81
  82     @Override
  83     public void setQueryParams(final Instant startTime, final Instant stopTime, final String filter,
  84                                final boolean leafNodesOnly, final boolean skipNotificationData) {
  85         super.setQueryParams(startTime, stopTime, filter, leafNodesOnly, skipNotificationData);
  86         try {
  87             this.formatter = getFormatter(filter);
  88         } catch (XPathExpressionException e) {
  89             throw new IllegalArgumentException("Failed to get filter", e);
  90         }
  91     }
  92
  93     /**
  94      * Get output type of this listener.
  95      *
  96      * @return The configured output type (JSON or XML).
  97      */
  98     @Override
  99     public String getOutputType() {
 100         return this.outputType.getName();
 101     }
 102
 103     @Override
 104     @SuppressWarnings("checkstyle:IllegalCatch")
 105     public void onNotification(final DOMNotification notification) {
 106         final Instant now = Instant.now();
 107         if (!checkStartStop(now, this)) {
 108             return;
 109         }
 110
 111         final Optional<String> maybeOutput;
 112         try {
 113             maybeOutput = formatter.eventData(schemaHandler.get(), notification, now, getLeafNodesOnly(),
 114                     isSkipNotificationData());
 115         } catch (Exception e) {
 116             LOG.error("Failed to process notification {}", notification, e);
 117             return;
 118         }
 119         if (maybeOutput.isPresent()) {
 120             post(maybeOutput.get());
 121         }
 122     }
 123
 124     /**
 125      * Get stream name of this listener.
 126      *
 127      * @return The configured stream name.
 128      */
 129     @Override
 130     public String getStreamName() {
 131         return this.streamName;
 132     }
 133
 134     /**
 135      * Get schema path of notification.
 136      *
 137      * @return The configured schema path that points to observing YANG notification schema node.
 138      */
 139     public Absolute getSchemaPath() {
 140         return this.path;
 141     }
 142
 143     public final synchronized void listen(final DOMNotificationService notificationService) {
 144         if (!isListening()) {
 145             setRegistration(notificationService.registerNotificationListener(this, getSchemaPath()));
 146         }
 147     }
 148
 149     @Override
 150     public String toString() {
 151         return MoreObjects.toStringHelper(this)
 152                 .add("path", path)
 153                 .add("stream-name", streamName)
 154                 .add("output-type", outputType)
 155                 .toString();
 156     }
 157 }