restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/listeners/AbstractCommonSubscriber.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 com.google.common.base.Preconditions.checkState;
  12 import static java.util.Objects.requireNonNull;
  13
  14 import com.google.common.base.MoreObjects;
  15 import java.time.Instant;
  16 import java.time.format.DateTimeFormatter;
  17 import java.time.format.DateTimeFormatterBuilder;
  18 import java.time.format.DateTimeParseException;
  19 import java.time.temporal.ChronoField;
  20 import java.time.temporal.TemporalAccessor;
  21 import java.util.HashSet;
  22 import java.util.Iterator;
  23 import java.util.Set;
  24 import java.util.concurrent.ExecutionException;
  25 import javax.xml.xpath.XPathExpressionException;
  26 import org.checkerframework.checker.lock.qual.GuardedBy;
  27 import org.checkerframework.checker.lock.qual.Holding;
  28 import org.eclipse.jdt.annotation.NonNull;
  29 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  30 import org.opendaylight.restconf.common.formatters.EventFormatter;
  31 import org.opendaylight.restconf.common.formatters.EventFormatterFactory;
  32 import org.opendaylight.restconf.nb.rfc8040.NotificationQueryParams;
  33 import org.opendaylight.restconf.nb.rfc8040.streams.StreamSessionHandler;
  34 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  35 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  36 import org.opendaylight.yangtools.concepts.Registration;
  37 import org.opendaylight.yangtools.yang.common.QName;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 /**
  42  * Features of subscribing part of both notifications.
  43  */
  44 abstract class AbstractCommonSubscriber<P, T> extends AbstractNotificationsData implements BaseListenerInterface {
  45     private static final Logger LOG = LoggerFactory.getLogger(AbstractCommonSubscriber.class);
  46     private static final DateTimeFormatter FORMATTER = new DateTimeFormatterBuilder()
  47         .appendValue(ChronoField.YEAR, 4).appendLiteral('-')
  48         .appendValue(ChronoField.MONTH_OF_YEAR, 2).appendLiteral('-')
  49         .appendValue(ChronoField.DAY_OF_MONTH, 2).appendLiteral('T')
  50         .appendValue(ChronoField.HOUR_OF_DAY, 2).appendLiteral(':')
  51         .appendValue(ChronoField.MINUTE_OF_HOUR, 2).appendLiteral(':')
  52         .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
  53         .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
  54         .appendOffset("+HH:MM", "Z").toFormatter();
  55
  56     private final EventFormatterFactory<T> formatterFactory;
  57     private final NotificationOutputType outputType;
  58     private final String streamName;
  59     private final P path;
  60
  61     @GuardedBy("this")
  62     private final Set<StreamSessionHandler> subscribers = new HashSet<>();
  63     @GuardedBy("this")
  64     private Registration registration;
  65
  66     // FIXME: these should be final
  67     private Instant start = null;
  68     private Instant stop = null;
  69     private boolean leafNodesOnly = false;
  70     private boolean skipNotificationData = false;
  71     private EventFormatter<T> formatter;
  72
  73     AbstractCommonSubscriber(final QName lastQName, final String streamName, final P path,
  74             final NotificationOutputType outputType, final EventFormatterFactory<T> formatterFactory) {
  75         super(lastQName);
  76         this.streamName = requireNonNull(streamName);
  77         checkArgument(!streamName.isEmpty());
  78         this.path = requireNonNull(path);
  79
  80         this.outputType = requireNonNull(outputType);
  81         this.formatterFactory = requireNonNull(formatterFactory);
  82         formatter = formatterFactory.getFormatter();
  83     }
  84
  85     @Override
  86     public final String getStreamName() {
  87         return streamName;
  88     }
  89
  90     @Override
  91     public final String getOutputType() {
  92         return outputType.getName();
  93     }
  94
  95     @Override
  96     public final synchronized boolean hasSubscribers() {
  97         return !subscribers.isEmpty();
  98     }
  99
 100     @Override
 101     public final synchronized Set<StreamSessionHandler> getSubscribers() {
 102         return new HashSet<>(subscribers);
 103     }
 104
 105     @Override
 106     public final synchronized void close() throws InterruptedException, ExecutionException {
 107         if (registration != null) {
 108             registration.close();
 109             registration = null;
 110         }
 111         deleteDataInDS().get();
 112         subscribers.clear();
 113     }
 114
 115     @Override
 116     public synchronized void addSubscriber(final StreamSessionHandler subscriber) {
 117         final boolean isConnected = subscriber.isConnected();
 118         checkState(isConnected);
 119         LOG.debug("Subscriber {} is added.", subscriber);
 120         subscribers.add(subscriber);
 121     }
 122
 123     @Override
 124     public synchronized void removeSubscriber(final StreamSessionHandler subscriber) {
 125         final boolean isConnected = subscriber.isConnected();
 126         checkState(isConnected);
 127         LOG.debug("Subscriber {} is removed", subscriber);
 128         subscribers.remove(subscriber);
 129         if (!hasSubscribers()) {
 130             ListenersBroker.getInstance().removeAndCloseListener(this);
 131         }
 132     }
 133
 134     public final Instant getStart() {
 135         return start;
 136     }
 137
 138     /**
 139      * Set query parameters for listener.
 140      *
 141      * @param params NotificationQueryParams to use.
 142      */
 143     public final void setQueryParams(final NotificationQueryParams params) {
 144         final var startTime = params.startTime();
 145         start = startTime == null ? Instant.now() : parseDateAndTime(startTime.value());
 146
 147         final var stopTime = params.stopTime();
 148         stop = stopTime == null ? null : parseDateAndTime(stopTime.value());
 149
 150         final var leafNodes = params.leafNodesOnly();
 151         leafNodesOnly = leafNodes == null ? false : leafNodes.value();
 152
 153         final var skipData = params.skipNotificationData();
 154         skipNotificationData = skipData == null ? false : skipData.value();
 155
 156         final var filter = params.filter();
 157         final String filterValue = filter == null ? null : filter.paramValue();
 158         if (filterValue != null && !filterValue.isEmpty()) {
 159             try {
 160                 formatter = formatterFactory.getFormatter(filterValue);
 161             } catch (XPathExpressionException e) {
 162                 throw new IllegalArgumentException("Failed to get filter", e);
 163             }
 164         } else {
 165             formatter = formatterFactory.getFormatter();
 166         }
 167     }
 168
 169     final P path() {
 170         return path;
 171     }
 172
 173     /**
 174      * Check whether this query should only notify about leaf node changes.
 175      *
 176      * @return true if this query should only notify about leaf node changes
 177      */
 178     final boolean getLeafNodesOnly() {
 179         return leafNodesOnly;
 180     }
 181
 182     /**
 183      * Check whether this query should notify changes without data.
 184      *
 185      * @return true if this query should notify about changes with  data
 186      */
 187     final boolean isSkipNotificationData() {
 188         return skipNotificationData;
 189     }
 190
 191     final EventFormatter<T> formatter() {
 192         return formatter;
 193     }
 194
 195     /**
 196      * Sets {@link Registration} registration.
 197      *
 198      * @param registration a listener registration registration.
 199      */
 200     @Holding("this")
 201     final void setRegistration(final Registration registration) {
 202         this.registration = requireNonNull(registration);
 203     }
 204
 205     /**
 206      * Checks if {@link Registration} registration exists.
 207      *
 208      * @return {@code true} if exists, {@code false} otherwise.
 209      */
 210     @Holding("this")
 211     final boolean isListening() {
 212         return registration != null;
 213     }
 214
 215     /**
 216      * Post data to subscribed SSE session handlers.
 217      *
 218      * @param data Data of incoming notifications.
 219      */
 220     synchronized void post(final String data) {
 221         final Iterator<StreamSessionHandler> iterator = subscribers.iterator();
 222         while (iterator.hasNext()) {
 223             final StreamSessionHandler subscriber = iterator.next();
 224             final boolean isConnected = subscriber.isConnected();
 225             if (isConnected) {
 226                 subscriber.sendDataMessage(data);
 227                 LOG.debug("Data was sent to subscriber {} on connection {}:", this, subscriber);
 228             } else {
 229                 // removal is probably not necessary, because it will be removed explicitly soon after invocation of
 230                 // onWebSocketClosed(..) in handler; but just to be sure ...
 231                 iterator.remove();
 232                 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);
 233             }
 234         }
 235     }
 236
 237     @SuppressWarnings("checkstyle:IllegalCatch")
 238     final boolean checkStartStop(final Instant now) {
 239         if (stop != null) {
 240             if (start.compareTo(now) < 0 && stop.compareTo(now) > 0) {
 241                 return true;
 242             }
 243             if (stop.compareTo(now) < 0) {
 244                 try {
 245                     close();
 246                 } catch (final Exception e) {
 247                     throw new RestconfDocumentedException("Problem with unregister listener." + e);
 248                 }
 249             }
 250         } else if (start != null) {
 251             if (start.compareTo(now) < 0) {
 252                 start = null;
 253                 return true;
 254             }
 255         } else {
 256             return true;
 257         }
 258         return false;
 259     }
 260
 261     @Override
 262     public final String toString() {
 263         return MoreObjects.toStringHelper(this)
 264             .add("path", path)
 265             .add("stream-name", streamName)
 266             .add("output-type", getOutputType())
 267             .toString();
 268     }
 269
 270     /**
 271      * Parse input of query parameters - start-time or stop-time - from {@link DateAndTime} format
 272      * to {@link Instant} format.
 273      *
 274      * @param uriValue Start-time or stop-time as string in {@link DateAndTime} format.
 275      * @return Parsed {@link Instant} by entry.
 276      */
 277     private static @NonNull Instant parseDateAndTime(final DateAndTime dateAndTime) {
 278         final TemporalAccessor accessor;
 279         try {
 280             accessor = FORMATTER.parse(dateAndTime.getValue());
 281         } catch (final DateTimeParseException e) {
 282             throw new RestconfDocumentedException("Cannot parse of value in date: " + dateAndTime, e);
 283         }
 284         return Instant.from(accessor);
 285     }
 286 }