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.net.InetSocketAddress;
  12 import java.util.HashSet;
  13 import java.util.Optional;
  14 import java.util.Set;
  15 import org.opendaylight.restconf.nb.rfc8040.streams.websockets.WebSocketSessionHandler;
  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<WebSocketSessionHandler> 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<WebSocketSessionHandler> getSubscribers() {
  37         return new HashSet<>(this.subscribers);
  38     }
  39
  40     @Override
  41     public final synchronized void close() throws Exception {
  42         if (this.registration != null) {
  43             this.registration.close();
  44             this.registration = null;
  45         }
  46         deleteDataInDS();
  47         this.subscribers.clear();
  48     }
  49
  50     @Override
  51     public synchronized void addSubscriber(final WebSocketSessionHandler subscriber) {
  52         final Optional<InetSocketAddress> remoteEndpointAddress = subscriber.getRemoteEndpointAddress();
  53         Preconditions.checkState(remoteEndpointAddress.isPresent());
  54         LOG.debug("Subscriber {} is added.", remoteEndpointAddress.get());
  55         subscribers.add(subscriber);
  56     }
  57
  58     @Override
  59     public synchronized void removeSubscriber(final WebSocketSessionHandler subscriber) {
  60         final Optional<InetSocketAddress> remoteEndpointAddress = subscriber.getRemoteEndpointAddress();
  61         Preconditions.checkState(remoteEndpointAddress.isPresent());
  62         LOG.debug("Subscriber {} is removed.", remoteEndpointAddress.get());
  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 web-socket session handlers.
  81      *
  82      * @param data Data of incoming notifications.
  83      */
  84     synchronized void post(final String data) {
  85         for (final WebSocketSessionHandler subscriber : subscribers) {
  86             final Optional<InetSocketAddress> remoteEndpointAddress = subscriber.getRemoteEndpointAddress();
  87             if (remoteEndpointAddress.isPresent()) {
  88                 subscriber.sendDataMessage(data);
  89                 LOG.debug("Data was sent to subscriber {} on address {}:", this, remoteEndpointAddress.get());
  90             } else {
  91                 // removal is probably not necessary, because it will be removed explicitly soon after invocation of
  92                 // onWebSocketClosed(..) in handler; but just to be sure ...
  93                 subscribers.remove(subscriber);
  94                 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);
  95             }
  96         }
  97     }
  98 }