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.opendaylight.yangtools.concepts.AbstractRegistration;
  31 import org.opendaylight.yangtools.concepts.Registration;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 /**
  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.
  39  */
  40 class NetconfSessionMonitoringService implements SessionListener, AutoCloseable {
  41     private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionMonitoringService.class);
  42
  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;
  49
  50     /**
  51      * Constructor for {@code NetconfSessionMonitoringService}.
  52      *
  53      * @param schedulingThreadPool thread pool for scheduling session stats updates. If not present, updates won't be
  54      *                             scheduled.
  55      * @param updateInterval       update interval. If is less than 0, updates won't be scheduled
  56      */
  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);
  63         } else {
  64             LOG.info("Scheduling thread pool is present = {}, "
  65                     + "update interval {}: /netconf-state/sessions won't be updated.",
  66                     schedulingThreadPool.isPresent(), updateInterval);
  67             executor = null;
  68         }
  69     }
  70
  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))
  76                 .build();
  77     }
  78
  79     @Override
  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);
  85     }
  86
  87     @Override
  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);
  94     }
  95
  96     @Override
  97     public synchronized void onSessionEvent(final SessionEvent event) {
  98         changedSessions.add(event.getSession());
  99     }
 100
 101     synchronized Registration registerListener(final NetconfMonitoringService.SessionsListener listener) {
 102         listeners.add(listener);
 103         if (!running) {
 104             startUpdateSessionStats();
 105         }
 106         return new AbstractRegistration() {
 107             @Override
 108             protected void removeRegistration() {
 109                 listeners.remove(listener);
 110             }
 111         };
 112     }
 113
 114     @Override
 115     public synchronized void close() {
 116         stopUpdateSessionStats();
 117         listeners.clear();
 118         sessions.clear();
 119     }
 120
 121     private synchronized void updateSessionStats() {
 122         if (changedSessions.isEmpty()) {
 123             return;
 124         }
 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);
 131         }
 132         changedSessions.clear();
 133     }
 134
 135     private void notifySessionUp(final NetconfManagementSession managementSession) {
 136         Session session = managementSession.toManagementSession();
 137         for (NetconfMonitoringService.SessionsListener listener : listeners) {
 138             listener.onSessionStarted(session);
 139         }
 140     }
 141
 142     private void notifySessionDown(final NetconfManagementSession managementSession) {
 143         Session session = managementSession.toManagementSession();
 144         for (NetconfMonitoringService.SessionsListener listener : listeners) {
 145             listener.onSessionEnded(session);
 146         }
 147     }
 148
 149     private void startUpdateSessionStats() {
 150         if (executor != null) {
 151             executor.scheduleAtFixedRate(this::updateSessionStats, 1, updateInterval, TimeUnit.SECONDS);
 152             running = true;
 153         }
 154     }
 155
 156     private void stopUpdateSessionStats() {
 157         if (executor != null) {
 158             executor.shutdownNow();
 159             running = false;
 160         }
 161     }
 162 }