Move netconf{-client,impl,util} to protocol/
[netconf.git] / protocol / netconf-impl / src / main / java / org / opendaylight / netconf / impl / osgi / NetconfSessionMonitoringService.java
1 /*
2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.netconf.impl.osgi;
9
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;
18 import java.util.Set;
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.api.monitoring.NetconfManagementSession;
24 import org.opendaylight.netconf.api.monitoring.NetconfMonitoringService;
25 import org.opendaylight.netconf.api.monitoring.SessionEvent;
26 import org.opendaylight.netconf.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.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
32
33 /**
34  * This class implements {@link SessionListener} to receive updates about Netconf sessions. Instance notifies its
35  * listeners about session start and end. It also publishes on regular interval list of sessions,
36  * where events like rpc or notification happened.
37  */
38 class NetconfSessionMonitoringService implements SessionListener, AutoCloseable {
39     private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionMonitoringService.class);
40
41     private final Set<NetconfManagementSession> sessions = new HashSet<>();
42     private final Set<NetconfManagementSession> changedSessions = new HashSet<>();
43     private final Set<NetconfMonitoringService.SessionsListener> listeners = new HashSet<>();
44     private final ScheduledExecutorService executor;
45     private final long updateInterval;
46     private boolean running;
47
48     /**
49      * Constructor for {@code NetconfSessionMonitoringService}.
50      *
51      * @param schedulingThreadPool thread pool for scheduling session stats updates. If not present, updates won't be
52      *                             scheduled.
53      * @param updateInterval       update interval. If is less than 0, updates won't be scheduled
54      */
55     NetconfSessionMonitoringService(final Optional<ScheduledThreadPool> schedulingThreadPool,
56             final long updateInterval) {
57         this.updateInterval = updateInterval;
58         if (schedulingThreadPool.isPresent() && updateInterval > 0) {
59             this.executor = schedulingThreadPool.get().getExecutor();
60             LOG.info("/netconf-state/sessions will be updated every {} seconds.", updateInterval);
61         } else {
62             LOG.info("Scheduling thread pool is present = {}, "
63                     + "update interval {}: /netconf-state/sessions won't be updated.",
64                     schedulingThreadPool.isPresent(), updateInterval);
65             this.executor = null;
66         }
67     }
68
69     synchronized Sessions getSessions() {
70         final Collection<Session> managementSessions = Collections2.transform(sessions,
71                 NetconfManagementSession::toManagementSession);
72         return new SessionsBuilder()
73                 .setSession(Maps.uniqueIndex(managementSessions, Session::key))
74                 .build();
75     }
76
77     @Override
78     public synchronized void onSessionUp(final NetconfManagementSession session) {
79         LOG.debug("Session {} up", session);
80         Preconditions.checkState(!sessions.contains(session), "Session %s was already added", session);
81         sessions.add(session);
82         notifySessionUp(session);
83     }
84
85     @Override
86     public synchronized void onSessionDown(final NetconfManagementSession session) {
87         LOG.debug("Session {} down", session);
88         Preconditions.checkState(sessions.contains(session), "Session %s not present", session);
89         sessions.remove(session);
90         changedSessions.remove(session);
91         notifySessionDown(session);
92     }
93
94     @Override
95     public synchronized void onSessionEvent(final SessionEvent event) {
96         changedSessions.add(event.getSession());
97     }
98
99     synchronized AutoCloseable registerListener(final NetconfMonitoringService.SessionsListener listener) {
100         listeners.add(listener);
101         if (!running) {
102             startUpdateSessionStats();
103         }
104         return () -> listeners.remove(listener);
105     }
106
107     @Override
108     public synchronized void close() {
109         stopUpdateSessionStats();
110         listeners.clear();
111         sessions.clear();
112     }
113
114     private synchronized void updateSessionStats() {
115         if (changedSessions.isEmpty()) {
116             return;
117         }
118         final List<Session> changed = changedSessions.stream()
119                 .map(NetconfManagementSession::toManagementSession)
120                 .collect(Collectors.toList());
121         final ImmutableList<Session> sessionImmutableList = ImmutableList.copyOf(changed);
122         for (NetconfMonitoringService.SessionsListener listener : listeners) {
123             listener.onSessionsUpdated(sessionImmutableList);
124         }
125         changedSessions.clear();
126     }
127
128     private void notifySessionUp(final NetconfManagementSession managementSession) {
129         Session session = managementSession.toManagementSession();
130         for (NetconfMonitoringService.SessionsListener listener : listeners) {
131             listener.onSessionStarted(session);
132         }
133     }
134
135     private void notifySessionDown(final NetconfManagementSession managementSession) {
136         Session session = managementSession.toManagementSession();
137         for (NetconfMonitoringService.SessionsListener listener : listeners) {
138             listener.onSessionEnded(session);
139         }
140     }
141
142     private void startUpdateSessionStats() {
143         if (executor != null) {
144             executor.scheduleAtFixedRate(this::updateSessionStats, 1, updateInterval, TimeUnit.SECONDS);
145             running = true;
146         }
147     }
148
149     private void stopUpdateSessionStats() {
150         if (executor != null) {
151             executor.shutdownNow();
152             running = false;
153         }
154     }
155 }