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 com.google.common.base.Preconditions;
  11 import java.util.HashSet;
  12 import java.util.Iterator;
  13 import java.util.Set;
  14 import java.util.concurrent.ExecutionException;
  15 import org.opendaylight.restconf.nb.rfc8040.streams.StreamSessionHandler;
  16 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  17 import org.slf4j.Logger;
  18 import org.slf4j.LoggerFactory;
  19
  20 /**
  21  * Features of subscribing part of both notifications.
  22  */
  23 abstract class AbstractCommonSubscriber extends AbstractQueryParams implements BaseListenerInterface {
  24
  25     private static final Logger LOG = LoggerFactory.getLogger(AbstractCommonSubscriber.class);
  26
  27     private final Set<StreamSessionHandler> subscribers = new HashSet<>();
  28     private volatile ListenerRegistration<?> registration;
  29
  30     @Override
  31     public final synchronized boolean hasSubscribers() {
  32         return !this.subscribers.isEmpty();
  33     }
  34
  35     @Override
  36     public final synchronized Set<StreamSessionHandler> getSubscribers() {
  37         return new HashSet<>(this.subscribers);
  38     }
  39
  40     @Override
  41     public final synchronized void close() throws InterruptedException, ExecutionException {
  42         if (this.registration != null) {
  43             this.registration.close();
  44             this.registration = null;
  45         }
  46         deleteDataInDS().get();
  47         this.subscribers.clear();
  48     }
  49
  50     @Override
  51     public synchronized void addSubscriber(final StreamSessionHandler subscriber) {
  52         final boolean isConnected = subscriber.isConnected();
  53         Preconditions.checkState(isConnected);
  54         LOG.debug("Subscriber {} is added.", subscriber);
  55         subscribers.add(subscriber);
  56     }
  57
  58     @Override
  59     public synchronized void removeSubscriber(final StreamSessionHandler subscriber) {
  60         final boolean isConnected = subscriber.isConnected();
  61         Preconditions.checkState(isConnected);
  62         LOG.debug("Subscriber {} is removed", subscriber);
  63         subscribers.remove(subscriber);
  64         if (!hasSubscribers()) {
  65             ListenersBroker.getInstance().removeAndCloseListener(this);
  66         }
  67     }
  68
  69     @Override
  70     public void setRegistration(final ListenerRegistration<?> registration) {
  71         this.registration = registration;
  72     }
  73
  74     @Override
  75     public boolean isListening() {
  76         return this.registration != null;
  77     }
  78
  79     /**
  80      * Post data to subscribed SSE session handlers.
  81      *
  82      * @param data Data of incoming notifications.
  83      */
  84     synchronized void post(final String data) {
  85         final Iterator<StreamSessionHandler> iterator = subscribers.iterator();
  86         while (iterator.hasNext()) {
  87             final StreamSessionHandler subscriber = iterator.next();
  88             final boolean isConnected = subscriber.isConnected();
  89             if (isConnected) {
  90                 subscriber.sendDataMessage(data);
  91                 LOG.debug("Data was sent to subscriber {} on connection {}:", this, subscriber);
  92             } else {
  93                 // removal is probably not necessary, because it will be removed explicitly soon after invocation of
  94                 // onWebSocketClosed(..) in handler; but just to be sure ...
  95                 iterator.remove();
  96                 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);
  97             }
  98         }
  99     }
 100 }