+++ /dev/null
-/*
- * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.netconf.impl.osgi;
-
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Collections2;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Sets;
-import java.util.Collection;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
-import org.opendaylight.netconf.api.monitoring.NetconfManagementSession;
-import org.opendaylight.netconf.api.monitoring.NetconfMonitoringService;
-import org.opendaylight.netconf.api.monitoring.SessionEvent;
-import org.opendaylight.netconf.api.monitoring.SessionListener;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Sessions;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.SessionsBuilder;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.sessions.Session;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class implements {@link SessionListener} to receive updates about Netconf sessions. Instance notifies its
- * listeners
- * about session start and end. It also publishes on regular interval list of sessions,
- * where events like rpc or notification happened.
- */
-class NetconfSessionMonitoringService implements SessionListener, AutoCloseable {
-
- private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionMonitoringService.class);
-
- private final Set<NetconfManagementSession> sessions = Sets.newHashSet();
- private final Set<NetconfManagementSession> changedSessions = Sets.newHashSet();
- private final Set<NetconfMonitoringService.SessionsListener> listeners = Sets.newHashSet();
- private final ScheduledExecutorService executor;
- private final long updateInterval;
- private boolean running;
-
- /**
- * Constructor for {@code NetconfSessionMonitoringService}.
- *
- * @param schedulingThreadPool thread pool for scheduling session stats updates. If not present, updates won't be
- * scheduled.
- * @param updateInterval update interval. If is less than 0, updates won't be scheduled
- */
- NetconfSessionMonitoringService(Optional<ScheduledThreadPool> schedulingThreadPool, long updateInterval) {
- this.updateInterval = updateInterval;
- if (schedulingThreadPool.isPresent() && updateInterval > 0) {
- this.executor = schedulingThreadPool.get().getExecutor();
- LOG.info("/netconf-state/sessions will be updated every {} seconds.", updateInterval);
- } else {
- LOG.info("Scheduling thread pool is present = {}, "
- + "update interval {}: /netconf-state/sessions won't be updated.",
- schedulingThreadPool.isPresent(), updateInterval);
- this.executor = null;
- }
- }
-
- synchronized Sessions getSessions() {
- final Collection<Session> managementSessions = Collections2.transform(sessions,
- NetconfManagementSession::toManagementSession);
- return new SessionsBuilder().setSession(ImmutableList.copyOf(managementSessions)).build();
- }
-
- @Override
- public synchronized void onSessionUp(final NetconfManagementSession session) {
- LOG.debug("Session {} up", session);
- Preconditions.checkState(!sessions.contains(session), "Session %s was already added", session);
- sessions.add(session);
- notifySessionUp(session);
- }
-
- @Override
- public synchronized void onSessionDown(final NetconfManagementSession session) {
- LOG.debug("Session {} down", session);
- Preconditions.checkState(sessions.contains(session), "Session %s not present", session);
- sessions.remove(session);
- changedSessions.remove(session);
- notifySessionDown(session);
- }
-
- @Override
- public synchronized void onSessionEvent(SessionEvent event) {
- changedSessions.add(event.getSession());
- }
-
- synchronized AutoCloseable registerListener(final NetconfMonitoringService.SessionsListener listener) {
- listeners.add(listener);
- if (!running) {
- startUpdateSessionStats();
- }
- return () -> listeners.remove(listener);
- }
-
- @Override
- public synchronized void close() {
- stopUpdateSessionStats();
- listeners.clear();
- sessions.clear();
- }
-
- private synchronized void updateSessionStats() {
- if (changedSessions.isEmpty()) {
- return;
- }
- final List<Session> changed = changedSessions.stream()
- .map(NetconfManagementSession::toManagementSession)
- .collect(Collectors.toList());
- final ImmutableList<Session> sessionImmutableList = ImmutableList.copyOf(changed);
- for (NetconfMonitoringService.SessionsListener listener : listeners) {
- listener.onSessionsUpdated(sessionImmutableList);
- }
- changedSessions.clear();
- }
-
- private void notifySessionUp(NetconfManagementSession managementSession) {
- Session session = managementSession.toManagementSession();
- for (NetconfMonitoringService.SessionsListener listener : listeners) {
- listener.onSessionStarted(session);
- }
- }
-
- private void notifySessionDown(NetconfManagementSession managementSession) {
- Session session = managementSession.toManagementSession();
- for (NetconfMonitoringService.SessionsListener listener : listeners) {
- listener.onSessionEnded(session);
- }
- }
-
- private void startUpdateSessionStats() {
- if (executor != null) {
- executor.scheduleAtFixedRate(this::updateSessionStats, 1, updateInterval, TimeUnit.SECONDS);
- running = true;
- }
- }
-
- private void stopUpdateSessionStats() {
- if (executor != null) {
- executor.shutdownNow();
- running = false;
- }
- }
-}
Code Review / transportpce.git / blobdiff