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.netconf.sal.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;
32 @SuppressWarnings("rawtypes")
33 private 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() {
54 this.registration.close();
55 this.registration = null;
60 * Creates event of type {@link EventType#REGISTER}, set {@link Channel}
61 * subscriber to the event and post event into event bus.
66 public void addSubscriber(final Channel subscriber) {
67 if (!subscriber.isActive()) {
68 LOG.debug("Channel is not active between websocket server and subscriber {}", subscriber.remoteAddress());
70 final Event event = new Event(EventType.REGISTER);
71 event.setSubscriber(subscriber);
72 this.eventBus.post(event);
76 * Creates event of type {@link EventType#DEREGISTER}, sets {@link Channel}
77 * subscriber to the event and posts event into event bus.
79 * @param subscriber subscriber channel
81 public void removeSubscriber(final Channel subscriber) {
82 LOG.debug("Subscriber {} is removed.", subscriber.remoteAddress());
83 final Event event = new Event(EventType.DEREGISTER);
84 event.setSubscriber(subscriber);
85 this.eventBus.post(event);
89 * Sets {@link ListenerRegistration} registration.
92 * DOMDataChangeListener registration
94 @SuppressWarnings("rawtypes")
95 public void setRegistration(final ListenerRegistration registration) {
96 this.registration = registration;
100 * Checks if {@link ListenerRegistration} registration exist.
102 * @return True if exist, false otherwise.
104 public boolean isListening() {
105 return this.registration != null;
109 * Creating and registering {@link EventBusChangeRecorder} of specific
110 * listener on {@link EventBus}.
113 * specific listener of notifications
115 @SuppressWarnings({ "unchecked", "rawtypes" })
116 protected <T extends BaseListenerInterface> void register(final T listener) {
117 this.eventBusChangeRecorder = new EventBusChangeRecorder(listener);
118 this.eventBus.register(this.eventBusChangeRecorder);
122 * Post event to event bus.
125 * data of incoming notifications
127 protected void post(final Event event) {
128 this.eventBus.post(event);
132 * Removes all subscribers and unregisters event bus change recorder form
135 protected void unregister() {
136 this.subscribers.clear();
137 this.eventBus.unregister(this.eventBusChangeRecorder);