2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.restconf.nb.rfc8040.streams.listeners;
10 import com.google.common.base.Preconditions;
11 import java.util.HashSet;
12 import java.util.Iterator;
14 import org.opendaylight.restconf.nb.rfc8040.streams.StreamSessionHandler;
15 import org.opendaylight.yangtools.concepts.ListenerRegistration;
16 import org.slf4j.Logger;
17 import org.slf4j.LoggerFactory;
20 * Features of subscribing part of both notifications.
22 abstract class AbstractCommonSubscriber extends AbstractQueryParams implements BaseListenerInterface {
24 private static final Logger LOG = LoggerFactory.getLogger(AbstractCommonSubscriber.class);
26 private final Set<StreamSessionHandler> subscribers = new HashSet<>();
27 private volatile ListenerRegistration<?> registration;
30 public final synchronized boolean hasSubscribers() {
31 return !this.subscribers.isEmpty();
35 public final synchronized Set<StreamSessionHandler> getSubscribers() {
36 return new HashSet<>(this.subscribers);
40 public final synchronized void close() throws Exception {
41 if (this.registration != null) {
42 this.registration.close();
43 this.registration = null;
46 this.subscribers.clear();
50 public synchronized void addSubscriber(final StreamSessionHandler subscriber) {
51 final boolean isConnected = subscriber.isConnected();
52 Preconditions.checkState(isConnected);
53 LOG.debug("Subscriber {} is added.", subscriber);
54 subscribers.add(subscriber);
58 public synchronized void removeSubscriber(final StreamSessionHandler subscriber) {
59 final boolean isConnected = subscriber.isConnected();
60 Preconditions.checkState(isConnected);
61 LOG.debug("Subscriber {} is removed", subscriber);
62 subscribers.remove(subscriber);
63 if (!hasSubscribers()) {
64 ListenersBroker.getInstance().removeAndCloseListener(this);
69 public void setRegistration(final ListenerRegistration<?> registration) {
70 this.registration = registration;
74 public boolean isListening() {
75 return this.registration != null;
79 * Post data to subscribed SSE session handlers.
81 * @param data Data of incoming notifications.
83 synchronized void post(final String data) {
84 final Iterator<StreamSessionHandler> iterator = subscribers.iterator();
85 while (iterator.hasNext()) {
86 final StreamSessionHandler subscriber = iterator.next();
87 final boolean isConnected = subscriber.isConnected();
89 subscriber.sendDataMessage(data);
90 LOG.debug("Data was sent to subscriber {} on connection {}:", this, subscriber);
92 // removal is probably not necessary, because it will be removed explicitly soon after invocation of
93 // onWebSocketClosed(..) in handler; but just to be sure ...
95 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);