import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableList.Builder;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import io.netty.util.internal.ConcurrentSet;
import java.util.ArrayList;
import java.util.Collection;
import org.opendaylight.controller.netconf.api.monitoring.NetconfMonitoringService;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Uri;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.NetconfState;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.NetconfStateBuilder;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.Yang;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Capabilities;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.CapabilitiesBuilder;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Schemas;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.SchemasBuilder;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.Sessions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class NetconfMonitoringServiceImpl implements NetconfMonitoringService {
+public class NetconfMonitoringServiceImpl implements NetconfMonitoringService, AutoCloseable {
private static final Schema.Location NETCONF_LOCATION = new Schema.Location(Schema.Location.Enumeration.NETCONF);
private static final List<Schema.Location> NETCONF_LOCATIONS = ImmutableList.of(NETCONF_LOCATION);
return input.toManagementSession();
}
};
-
- private final Set<NetconfManagementSession> sessions = new ConcurrentSet<>();
- private final NetconfOperationServiceFactory netconfOperationProvider;
- private final Map<String, Capability> capabilities = new ConcurrentHashMap<>();
- private static final Function<Capability, String> CAPABILITY_TO_URI = new Function<Capability, String>() {
+ private static final Function<Capability, Uri> CAPABILITY_TO_URI = new Function<Capability, Uri>() {
@Override
- public String apply(final Capability input) {
- return input.getCapabilityUri();
+ public Uri apply(final Capability input) {
+ return new Uri(input.getCapabilityUri());
}
};
- // FIXME check threadsafety
+ private final Set<NetconfManagementSession> sessions = new ConcurrentSet<>();
+ private final NetconfOperationServiceFactory netconfOperationProvider;
+ private final Map<Uri, Capability> capabilities = new ConcurrentHashMap<>();
+
+ private final Set<MonitoringListener> listeners = Sets.newHashSet();
public NetconfMonitoringServiceImpl(final NetconfOperationServiceFactory netconfOperationProvider) {
this.netconfOperationProvider = netconfOperationProvider;
}
@Override
- public void onSessionUp(final NetconfManagementSession session) {
+ 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);
+ notifyListeners();
}
@Override
- public void onSessionDown(final NetconfManagementSession session) {
+ 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);
+ notifyListeners();
}
@Override
- public Sessions getSessions() {
+ public synchronized Sessions getSessions() {
return new SessionsBuilder().setSession(ImmutableList.copyOf(Collections2.transform(sessions, SESSION_FUNCTION))).build();
}
@Override
- public Schemas getSchemas() {
+ public synchronized Schemas getSchemas() {
try {
return transformSchemas(netconfOperationProvider.getCapabilities());
} catch (final RuntimeException e) {
}
@Override
- public String getSchemaForCapability(final String moduleName, final Optional<String> revision) {
+ public synchronized String getSchemaForCapability(final String moduleName, final Optional<String> revision) {
// FIXME not effective at all
}
@Override
- public Set<String> getCapabilities() {
- return capabilities.keySet();
+ public synchronized Capabilities getCapabilities() {
+ return new CapabilitiesBuilder().setCapability(Lists.newArrayList(capabilities.keySet())).build();
+ }
+
+ @Override
+ public synchronized AutoCloseable registerListener(final MonitoringListener listener) {
+ listeners.add(listener);
+ listener.onStateChanged(getCurrentNetconfState());
+ return new AutoCloseable() {
+ @Override
+ public void close() throws Exception {
+ listeners.remove(listener);
+ }
+ };
+ }
+
+ private NetconfState getCurrentNetconfState() {
+ return new NetconfStateBuilder()
+ .setCapabilities(getCapabilities())
+ .setSchemas(getSchemas())
+ .setSessions(getSessions())
+ .build();
}
private static Schemas transformSchemas(final Set<Capability> caps) {
}
@Override
- public void onCapabilitiesAdded(final Set<Capability> addedCaps) {
- // FIXME what check for duplicates
+ public synchronized void onCapabilitiesAdded(final Set<Capability> addedCaps) {
+ // FIXME howto check for duplicates
this.capabilities.putAll(Maps.uniqueIndex(addedCaps, CAPABILITY_TO_URI));
+ notifyListeners();
+ }
+
+ private void notifyListeners() {
+ for (final MonitoringListener listener : listeners) {
+ listener.onStateChanged(getCurrentNetconfState());
+ }
}
@Override
- public void onCapabilitiesRemoved(final Set<Capability> addedCaps) {
+ public synchronized void onCapabilitiesRemoved(final Set<Capability> addedCaps) {
for (final Capability addedCap : addedCaps) {
capabilities.remove(addedCap.getCapabilityUri());
}
+ notifyListeners();
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ listeners.clear();
+ sessions.clear();
+ capabilities.clear();
}
}