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