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.eventbus.AsyncEventBus;
11 import com.google.common.eventbus.EventBus;
12 import io.netty.channel.Channel;
13 import io.netty.util.internal.ConcurrentSet;
15 import java.util.concurrent.Executors;
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<Channel> subscribers = new ConcurrentSet<>();
28 private final EventBus eventBus;
30 @SuppressWarnings("rawtypes")
31 private EventBusChangeRecorder eventBusChangeRecorder;
33 private volatile ListenerRegistration<?> registration;
36 * Creating {@link EventBus}.
38 protected AbstractCommonSubscriber() {
39 this.eventBus = new AsyncEventBus(Executors.newSingleThreadExecutor());
43 public final boolean hasSubscribers() {
44 return !this.subscribers.isEmpty();
48 public final Set<Channel> getSubscribers() {
49 return this.subscribers;
53 public final void close() throws Exception {
54 if (this.registration != null) {
55 this.registration.close();
56 this.registration = null;
64 * Creates event of type {@link EventType#REGISTER}, set {@link Channel}
65 * subscriber to the event and post event into event bus.
70 public void addSubscriber(final Channel subscriber) {
71 if (!subscriber.isActive()) {
72 LOG.debug("Channel is not active between websocket server and subscriber {}" + subscriber.remoteAddress());
74 final Event event = new Event(EventType.REGISTER);
75 event.setSubscriber(subscriber);
76 this.eventBus.post(event);
80 * Creates event of type {@link EventType#DEREGISTER}, sets {@link Channel}
81 * subscriber to the event and posts event into event bus.
83 * @param subscriber subscriber channel
85 public void removeSubscriber(final Channel subscriber) {
86 LOG.debug("Subscriber {} is removed.", subscriber.remoteAddress());
87 final Event event = new Event(EventType.DEREGISTER);
88 event.setSubscriber(subscriber);
89 this.eventBus.post(event);
93 * Sets {@link ListenerRegistration} registration.
96 * DOMDataChangeListener registration
98 public void setRegistration(final ListenerRegistration<?> registration) {
99 this.registration = registration;
103 * Checks if {@link ListenerRegistration} registration exist.
105 * @return True if exist, false otherwise.
107 public boolean isListening() {
108 return this.registration != null;
112 * Creating and registering {@link EventBusChangeRecorder} of specific
113 * listener on {@link EventBus}.
116 * specific listener of notifications
118 @SuppressWarnings({ "unchecked", "rawtypes" })
119 protected <T extends BaseListenerInterface> void register(final T listener) {
120 this.eventBusChangeRecorder = new EventBusChangeRecorder(listener);
121 this.eventBus.register(this.eventBusChangeRecorder);
125 * Post event to event bus.
128 * data of incoming notifications
130 protected void post(final Event event) {
131 this.eventBus.post(event);
135 * Removes all subscribers and unregisters event bus change recorder form
138 protected void unregister() {
139 this.subscribers.clear();
140 this.eventBus.unregister(this.eventBusChangeRecorder);