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 java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nonnull;
-import org.opendaylight.controller.netconf.api.Capability;
+import org.opendaylight.controller.config.util.capability.BasicCapability;
+import org.opendaylight.controller.config.util.capability.Capability;
import org.opendaylight.controller.netconf.api.monitoring.NetconfManagementSession;
import org.opendaylight.controller.netconf.api.monitoring.NetconfMonitoringService;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
+import org.opendaylight.controller.netconf.notifications.BaseNotificationPublisherRegistration;
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.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.schemas.SchemaBuilder;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.schemas.SchemaKey;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.sessions.Session;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChange;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChangeBuilder;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.changed.by.parms.ChangedByBuilder;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.changed.by.parms.changed.by.server.or.user.ServerBuilder;
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();
+ private volatile BaseNotificationPublisherRegistration notificationPublisher;
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) {
return b.build();
}
+ public static Set<Capability> setupCapabilities(final Set<Capability> caps) {
+ Set<Capability> capabilities = new HashSet<>(caps);
+ capabilities.add(new BasicCapability("urn:ietf:params:netconf:capability:candidate:1.0"));
+ // TODO rollback on error not supported EditConfigXmlParser:100
+ // [RFC6241] 8.5. Rollback-on-Error Capability
+ // capabilities.add(new BasicCapability("urn:ietf:params:netconf:capability:rollback-on-error:1.0"));
+ return capabilities;
+ }
+
@Override
- public void onCapabilitiesAdded(final Set<Capability> addedCaps) {
- // FIXME what check for duplicates
- this.capabilities.putAll(Maps.uniqueIndex(addedCaps, CAPABILITY_TO_URI));
+ public synchronized void close() throws Exception {
+ listeners.clear();
+ sessions.clear();
+ capabilities.clear();
}
@Override
- public void onCapabilitiesRemoved(final Set<Capability> addedCaps) {
+ public void onCapabilitiesChanged(Set<Capability> added, Set<Capability> removed) {
+ onCapabilitiesAdded(added);
+ onCapabilitiesRemoved(removed);
+ notifyListeners();
+
+ // publish notification to notification collector about changed capabilities
+ if (notificationPublisher != null) {
+ notificationPublisher.onCapabilityChanged(computeDiff(added, removed));
+ }
+ }
+
+ static NetconfCapabilityChange computeDiff(final Set<Capability> removed, final Set<Capability> added) {
+ final NetconfCapabilityChangeBuilder netconfCapabilityChangeBuilder = new NetconfCapabilityChangeBuilder();
+ netconfCapabilityChangeBuilder.setChangedBy(new ChangedByBuilder().setServerOrUser(new ServerBuilder().setServer(true).build()).build());
+ netconfCapabilityChangeBuilder.setDeletedCapability(Lists.newArrayList(Collections2.transform(removed, CAPABILITY_TO_URI)));
+ netconfCapabilityChangeBuilder.setAddedCapability(Lists.newArrayList(Collections2.transform(added, CAPABILITY_TO_URI)));
+ // TODO modified should be computed ... but why ?
+ netconfCapabilityChangeBuilder.setModifiedCapability(Collections.<Uri>emptyList());
+ return netconfCapabilityChangeBuilder.build();
+ }
+
+
+ private synchronized void onCapabilitiesAdded(final Set<Capability> addedCaps) {
+ // FIXME howto check for duplicates
+ this.capabilities.putAll(Maps.uniqueIndex(setupCapabilities(addedCaps), CAPABILITY_TO_URI));
+ }
+
+ private void notifyListeners() {
+ for (final MonitoringListener listener : listeners) {
+ listener.onStateChanged(getCurrentNetconfState());
+ }
+ }
+
+ private synchronized void onCapabilitiesRemoved(final Set<Capability> addedCaps) {
for (final Capability addedCap : addedCaps) {
- capabilities.remove(addedCap.getCapabilityUri());
+ capabilities.remove(CAPABILITY_TO_URI.apply(addedCap));
}
}
+
+ public void setNotificationPublisher(final BaseNotificationPublisherRegistration notificationPublisher) {
+ this.notificationPublisher = notificationPublisher;
+ }
}
Code Review / controller.git / blobdiff