protocol/netconf-server/src/main/java/org/opendaylight/netconf/server/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.server.osgi;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.collect.ImmutableList;
  13 import java.util.HashSet;
  14 import java.util.Set;
  15 import java.util.concurrent.ScheduledExecutorService;
  16 import java.util.concurrent.ScheduledFuture;
  17 import java.util.concurrent.TimeUnit;
  18 import java.util.stream.Collectors;
  19 import org.checkerframework.checker.lock.qual.Holding;
  20 import org.opendaylight.netconf.server.api.monitoring.NetconfManagementSession;
  21 import org.opendaylight.netconf.server.api.monitoring.NetconfMonitoringService;
  22 import org.opendaylight.netconf.server.api.monitoring.SessionEvent;
  23 import org.opendaylight.netconf.server.api.monitoring.SessionListener;
  24 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Sessions;
  25 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.SessionsBuilder;
  26 import org.opendaylight.yangtools.concepts.AbstractRegistration;
  27 import org.opendaylight.yangtools.concepts.Registration;
  28 import org.opendaylight.yangtools.yang.binding.util.BindingMap;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * This class implements {@link SessionListener} to receive updates about Netconf sessions. Instance notifies its
  34  * listeners about session start and end. It also publishes on regular interval list of sessions,
  35  * where events like rpc or notification happened.
  36  */
  37 abstract sealed class NetconfSessionMonitoringService implements SessionListener, AutoCloseable {
  38     static final class WithoutUpdates extends NetconfSessionMonitoringService {
  39         WithoutUpdates() {
  40             LOG.info("/netconf-state/sessions will not be updated.");
  41         }
  42
  43         @Override
  44         void startTask() {
  45            // No-op
  46         }
  47
  48         @Override
  49         void stopTask() {
  50             // No-op
  51         }
  52     }
  53
  54     static final class WithUpdates extends NetconfSessionMonitoringService {
  55         private final ScheduledExecutorService executor;
  56         private final long period;
  57         private final TimeUnit timeUnit;
  58
  59         private ScheduledFuture<?> task;
  60
  61         WithUpdates(final ScheduledExecutorService executor, final long period, final TimeUnit timeUnit) {
  62             this.executor = requireNonNull(executor);
  63             this.period = period;
  64             this.timeUnit = requireNonNull(timeUnit);
  65             LOG.info("/netconf-state/sessions will be updated every {} {}.", period, timeUnit);
  66         }
  67
  68         @Override
  69         void startTask() {
  70             if (task == null) {
  71                 task = executor.scheduleAtFixedRate(this::updateSessionStats, period, period, timeUnit);
  72             }
  73         }
  74
  75         @Override
  76         void stopTask() {
  77             if (task != null) {
  78                 task.cancel(false);
  79                 task = null;
  80             }
  81         }
  82     }
  83
  84     private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionMonitoringService.class);
  85
  86     private final Set<NetconfManagementSession> sessions = new HashSet<>();
  87     private final Set<NetconfManagementSession> changedSessions = new HashSet<>();
  88     private final Set<NetconfMonitoringService.SessionsListener> listeners = new HashSet<>();
  89
  90     synchronized Sessions getSessions() {
  91         return new SessionsBuilder()
  92             .setSession(BindingMap.of(
  93                 sessions.stream().map(NetconfManagementSession::toManagementSession).collect(Collectors.toList())))
  94             .build();
  95     }
  96
  97     @Override
  98     public final synchronized void onSessionUp(final NetconfManagementSession session) {
  99         LOG.debug("Session {} up", session);
 100         if (!sessions.add(session)) {
 101             throw new IllegalStateException("Session " + session + " was already added");
 102         }
 103
 104         final var mgmt = session.toManagementSession();
 105         for (var listener : listeners) {
 106             listener.onSessionStarted(mgmt);
 107         }
 108     }
 109
 110     @Override
 111     public final synchronized void onSessionDown(final NetconfManagementSession session) {
 112         LOG.debug("Session {} down", session);
 113         if (!sessions.remove(session)) {
 114             throw new IllegalStateException("Session " + session + " not present");
 115         }
 116         changedSessions.remove(session);
 117
 118         final var mgmt = session.toManagementSession();
 119         for (var listener : listeners) {
 120             listener.onSessionEnded(mgmt);
 121         }
 122     }
 123
 124     @Override
 125     public final synchronized void onSessionEvent(final SessionEvent event) {
 126         changedSessions.add(event.getSession());
 127     }
 128
 129     final synchronized Registration registerListener(final NetconfMonitoringService.SessionsListener listener) {
 130         listeners.add(listener);
 131         startTask();
 132         return new AbstractRegistration() {
 133             @Override
 134             protected void removeRegistration() {
 135                 listeners.remove(listener);
 136             }
 137         };
 138     }
 139
 140     @Override
 141     public final synchronized void close() {
 142         stopTask();
 143         listeners.clear();
 144         sessions.clear();
 145     }
 146
 147     @Holding("this")
 148     abstract void startTask();
 149
 150     @Holding("this")
 151     abstract void stopTask();
 152
 153     final synchronized void updateSessionStats() {
 154         if (!changedSessions.isEmpty()) {
 155             final var updatedSessions = changedSessions.stream()
 156                 .map(NetconfManagementSession::toManagementSession)
 157                 .collect(ImmutableList.toImmutableList());
 158             changedSessions.clear();
 159
 160             for (var listener : listeners) {
 161                 listener.onSessionsUpdated(updatedSessions);
 162             }
 163         }
 164     }
 165 }