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 static java.util.Objects.requireNonNull;
12 import com.google.common.base.Preconditions;
13 import java.util.HashSet;
14 import java.util.Iterator;
16 import java.util.concurrent.ExecutionException;
17 import org.checkerframework.checker.lock.qual.GuardedBy;
18 import org.checkerframework.checker.lock.qual.Holding;
19 import org.opendaylight.restconf.nb.rfc8040.streams.StreamSessionHandler;
20 import org.opendaylight.yangtools.concepts.Registration;
21 import org.slf4j.Logger;
22 import org.slf4j.LoggerFactory;
25 * Features of subscribing part of both notifications.
27 abstract class AbstractCommonSubscriber extends AbstractQueryParams implements BaseListenerInterface {
28 private static final Logger LOG = LoggerFactory.getLogger(AbstractCommonSubscriber.class);
31 private final Set<StreamSessionHandler> subscribers = new HashSet<>();
33 private Registration registration;
36 public final synchronized boolean hasSubscribers() {
37 return !this.subscribers.isEmpty();
41 public final synchronized Set<StreamSessionHandler> getSubscribers() {
42 return new HashSet<>(this.subscribers);
46 public final synchronized void close() throws InterruptedException, ExecutionException {
47 if (this.registration != null) {
48 this.registration.close();
49 this.registration = null;
51 deleteDataInDS().get();
52 this.subscribers.clear();
56 public synchronized void addSubscriber(final StreamSessionHandler subscriber) {
57 final boolean isConnected = subscriber.isConnected();
58 Preconditions.checkState(isConnected);
59 LOG.debug("Subscriber {} is added.", subscriber);
60 subscribers.add(subscriber);
64 public synchronized void removeSubscriber(final StreamSessionHandler subscriber) {
65 final boolean isConnected = subscriber.isConnected();
66 Preconditions.checkState(isConnected);
67 LOG.debug("Subscriber {} is removed", subscriber);
68 subscribers.remove(subscriber);
69 if (!hasSubscribers()) {
70 ListenersBroker.getInstance().removeAndCloseListener(this);
75 * Sets {@link Registration} registration.
77 * @param registration a listener registration registration.
80 final void setRegistration(final Registration registration) {
81 this.registration = requireNonNull(registration);
85 * Checks if {@link Registration} registration exists.
87 * @return {@code true} if exists, {@code false} otherwise.
90 final boolean isListening() {
91 return registration != null;
95 * Post data to subscribed SSE session handlers.
97 * @param data Data of incoming notifications.
99 synchronized void post(final String data) {
100 final Iterator<StreamSessionHandler> iterator = subscribers.iterator();
101 while (iterator.hasNext()) {
102 final StreamSessionHandler subscriber = iterator.next();
103 final boolean isConnected = subscriber.isConnected();
105 subscriber.sendDataMessage(data);
106 LOG.debug("Data was sent to subscriber {} on connection {}:", this, subscriber);
108 // removal is probably not necessary, because it will be removed explicitly soon after invocation of
109 // onWebSocketClosed(..) in handler; but just to be sure ...
111 LOG.debug("Subscriber for {} was removed - web-socket session is not open.", this);