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.server.osgi;
10 import com.google.common.base.Preconditions;
11 import com.google.common.collect.Collections2;
12 import com.google.common.collect.ImmutableList;
13 import com.google.common.collect.Maps;
14 import java.util.Collection;
15 import java.util.HashSet;
16 import java.util.List;
17 import java.util.Optional;
19 import java.util.concurrent.ScheduledExecutorService;
20 import java.util.concurrent.TimeUnit;
21 import java.util.stream.Collectors;
22 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
23 import org.opendaylight.netconf.server.api.monitoring.NetconfManagementSession;
24 import org.opendaylight.netconf.server.api.monitoring.NetconfMonitoringService;
25 import org.opendaylight.netconf.server.api.monitoring.SessionEvent;
26 import org.opendaylight.netconf.server.api.monitoring.SessionListener;
27 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Sessions;
28 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.SessionsBuilder;
29 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.sessions.Session;
30 import org.opendaylight.yangtools.concepts.AbstractRegistration;
31 import org.opendaylight.yangtools.concepts.Registration;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
36 * This class implements {@link SessionListener} to receive updates about Netconf sessions. Instance notifies its
37 * listeners about session start and end. It also publishes on regular interval list of sessions,
38 * where events like rpc or notification happened.
40 class NetconfSessionMonitoringService implements SessionListener, AutoCloseable {
41 private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionMonitoringService.class);
43 private final Set<NetconfManagementSession> sessions = new HashSet<>();
44 private final Set<NetconfManagementSession> changedSessions = new HashSet<>();
45 private final Set<NetconfMonitoringService.SessionsListener> listeners = new HashSet<>();
46 private final ScheduledExecutorService executor;
47 private final long updateInterval;
48 private boolean running;
51 * Constructor for {@code NetconfSessionMonitoringService}.
53 * @param schedulingThreadPool thread pool for scheduling session stats updates. If not present, updates won't be
55 * @param updateInterval update interval. If is less than 0, updates won't be scheduled
57 NetconfSessionMonitoringService(final Optional<ScheduledThreadPool> schedulingThreadPool,
58 final long updateInterval) {
59 this.updateInterval = updateInterval;
60 if (schedulingThreadPool.isPresent() && updateInterval > 0) {
61 executor = schedulingThreadPool.orElseThrow().getExecutor();
62 LOG.info("/netconf-state/sessions will be updated every {} seconds.", updateInterval);
64 LOG.info("Scheduling thread pool is present = {}, "
65 + "update interval {}: /netconf-state/sessions won't be updated.",
66 schedulingThreadPool.isPresent(), updateInterval);
71 synchronized Sessions getSessions() {
72 final Collection<Session> managementSessions = Collections2.transform(sessions,
73 NetconfManagementSession::toManagementSession);
74 return new SessionsBuilder()
75 .setSession(Maps.uniqueIndex(managementSessions, Session::key))
80 public synchronized void onSessionUp(final NetconfManagementSession session) {
81 LOG.debug("Session {} up", session);
82 Preconditions.checkState(!sessions.contains(session), "Session %s was already added", session);
83 sessions.add(session);
84 notifySessionUp(session);
88 public synchronized void onSessionDown(final NetconfManagementSession session) {
89 LOG.debug("Session {} down", session);
90 Preconditions.checkState(sessions.contains(session), "Session %s not present", session);
91 sessions.remove(session);
92 changedSessions.remove(session);
93 notifySessionDown(session);
97 public synchronized void onSessionEvent(final SessionEvent event) {
98 changedSessions.add(event.getSession());
101 synchronized Registration registerListener(final NetconfMonitoringService.SessionsListener listener) {
102 listeners.add(listener);
104 startUpdateSessionStats();
106 return new AbstractRegistration() {
108 protected void removeRegistration() {
109 listeners.remove(listener);
115 public synchronized void close() {
116 stopUpdateSessionStats();
121 private synchronized void updateSessionStats() {
122 if (changedSessions.isEmpty()) {
125 final List<Session> changed = changedSessions.stream()
126 .map(NetconfManagementSession::toManagementSession)
127 .collect(Collectors.toList());
128 final ImmutableList<Session> sessionImmutableList = ImmutableList.copyOf(changed);
129 for (NetconfMonitoringService.SessionsListener listener : listeners) {
130 listener.onSessionsUpdated(sessionImmutableList);
132 changedSessions.clear();
135 private void notifySessionUp(final NetconfManagementSession managementSession) {
136 Session session = managementSession.toManagementSession();
137 for (NetconfMonitoringService.SessionsListener listener : listeners) {
138 listener.onSessionStarted(session);
142 private void notifySessionDown(final NetconfManagementSession managementSession) {
143 Session session = managementSession.toManagementSession();
144 for (NetconfMonitoringService.SessionsListener listener : listeners) {
145 listener.onSessionEnded(session);
149 private void startUpdateSessionStats() {
150 if (executor != null) {
151 executor.scheduleAtFixedRate(this::updateSessionStats, 1, updateInterval, TimeUnit.SECONDS);
156 private void stopUpdateSessionStats() {
157 if (executor != null) {
158 executor.shutdownNow();