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 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;
21 * Features of subscribing part of both notifications.
23 abstract class AbstractCommonSubscriber extends AbstractQueryParams implements BaseListenerInterface {
25 private static final Logger LOG = LoggerFactory.getLogger(AbstractCommonSubscriber.class);
27 private final Set<StreamSessionHandler> subscribers = new HashSet<>();
28 private volatile ListenerRegistration<?> registration;
31 public final synchronized boolean hasSubscribers() {
32 return !this.subscribers.isEmpty();
36 public final synchronized Set<StreamSessionHandler> getSubscribers() {
37 return new HashSet<>(this.subscribers);
41 public final synchronized void close() throws InterruptedException, ExecutionException {
42 if (this.registration != null) {
43 this.registration.close();
44 this.registration = null;
46 deleteDataInDS().get();
47 this.subscribers.clear();
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);
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);
70 public void setRegistration(final ListenerRegistration<?> registration) {
71 this.registration = registration;
75 public boolean isListening() {
76 return this.registration != null;
80 * Post data to subscribed SSE session handlers.
82 * @param data Data of incoming notifications.
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();
90 subscriber.sendDataMessage(data);
91 LOG.debug("Data was sent to subscriber {} on connection {}:", this, subscriber);
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 ...
96 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);