opendaylight/netconf/config-netconf-connector/src/main/java/org/opendaylight/controller/netconf/confignetconfconnector/osgi/YangStoreService.java

   1 /*
   2  * Copyright (c) 2013 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
   9 package org.opendaylight.controller.netconf.confignetconfconnector.osgi;
  10
  11 import com.google.common.base.Function;
  12 import com.google.common.collect.Collections2;
  13 import com.google.common.collect.Lists;
  14 import com.google.common.collect.Sets;
  15 import java.lang.ref.SoftReference;
  16 import java.util.Collections;
  17 import java.util.HashSet;
  18 import java.util.Map;
  19 import java.util.Set;
  20 import java.util.concurrent.ExecutorService;
  21 import java.util.concurrent.Executors;
  22 import java.util.concurrent.ThreadFactory;
  23 import java.util.concurrent.atomic.AtomicReference;
  24 import org.opendaylight.controller.config.yangjmxgenerator.ModuleMXBeanEntry;
  25 import org.opendaylight.controller.netconf.api.Capability;
  26 import org.opendaylight.controller.netconf.api.monitoring.CapabilityListener;
  27 import org.opendaylight.controller.netconf.notifications.BaseNetconfNotificationListener;
  28 import org.opendaylight.controller.netconf.notifications.BaseNotificationPublisherRegistration;
  29 import org.opendaylight.controller.netconf.notifications.NetconfNotificationCollector;
  30 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Uri;
  31 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChange;
  32 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChangeBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.changed.by.parms.ChangedByBuilder;
  34 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;
  35 import org.opendaylight.yangtools.yang.common.QName;
  36 import org.opendaylight.yangtools.yang.model.api.Module;
  37 import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
  38 import org.opendaylight.yangtools.yang.model.api.SchemaContextProvider;
  39 import org.osgi.framework.BundleContext;
  40 import org.osgi.framework.ServiceReference;
  41 import org.osgi.util.tracker.ServiceTracker;
  42 import org.osgi.util.tracker.ServiceTrackerCustomizer;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45
  46 public class YangStoreService implements YangStoreContext {
  47
  48     private static final Logger LOG = LoggerFactory.getLogger(YangStoreService.class);
  49
  50     /**
  51      * This is a rather interesting locking model. We need to guard against both the
  52      * cache expiring from GC and being invalidated by schema context change. The
  53      * context can change while we are doing processing, so we do not want to block
  54      * it, so no synchronization can happen on the methods.
  55      *
  56      * So what we are doing is the following:
  57      *
  58      * We synchronize with GC as usual, using a SoftReference.
  59      *
  60      * The atomic reference is used to synchronize with {@link #refresh()}, e.g. when
  61      * refresh happens, it will push a SoftReference(null), e.g. simulate the GC. Now
  62      * that may happen while the getter is already busy acting on the old schema context,
  63      * so it needs to understand that a refresh has happened and retry. To do that, it
  64      * attempts a CAS operation -- if it fails, in knows that the SoftReference has
  65      * been replaced and thus it needs to retry.
  66      *
  67      * Note that {@link #getYangStoreSnapshot()} will still use synchronize() internally
  68      * to stop multiple threads doing the same work.
  69      */
  70     private final AtomicReference<SoftReference<YangStoreSnapshot>> ref =
  71             new AtomicReference<>(new SoftReference<YangStoreSnapshot>(null));
  72
  73     private final SchemaContextProvider schemaContextProvider;
  74     private final BaseNetconfNotificationListener notificationPublisher;
  75
  76     private final ExecutorService notificationExecutor = Executors.newSingleThreadExecutor(new ThreadFactory() {
  77         @Override
  78         public Thread newThread(final Runnable r) {
  79             return new Thread(r, "config-netconf-connector-capability-notifications");
  80         }
  81     });
  82
  83     private final Set<CapabilityListener> listeners = Collections.synchronizedSet(new HashSet<CapabilityListener>());
  84
  85     public YangStoreService(final SchemaContextProvider schemaContextProvider, final BundleContext context) {
  86         this(schemaContextProvider, new NotificationCollectorTracker(context));
  87     }
  88
  89     public YangStoreService(final SchemaContextProvider schemaContextProvider, final BaseNetconfNotificationListener notificationHandler) {
  90         this.schemaContextProvider = schemaContextProvider;
  91         this.notificationPublisher = notificationHandler;
  92     }
  93
  94     private synchronized YangStoreContext getYangStoreSnapshot() {
  95         SoftReference<YangStoreSnapshot> r = ref.get();
  96         YangStoreSnapshot ret = r.get();
  97
  98         while (ret == null) {
  99             // We need to be compute a new value
 100             ret = new YangStoreSnapshot(schemaContextProvider.getSchemaContext());
 101
 102             if (!ref.compareAndSet(r, new SoftReference<>(ret))) {
 103                 LOG.debug("Concurrent refresh detected, recomputing snapshot");
 104                 r = ref.get();
 105                 ret = null;
 106             }
 107         }
 108
 109         return ret;
 110     }
 111
 112     @Override
 113     public Map<String, Map<String, ModuleMXBeanEntry>> getModuleMXBeanEntryMap() {
 114         return getYangStoreSnapshot().getModuleMXBeanEntryMap();
 115     }
 116
 117     @Override
 118     public Map<QName, Map<String, ModuleMXBeanEntry>> getQNamesToIdentitiesToModuleMXBeanEntries() {
 119         return getYangStoreSnapshot().getQNamesToIdentitiesToModuleMXBeanEntries();
 120     }
 121
 122     @Override
 123     public Set<Module> getModules() {
 124         return getYangStoreSnapshot().getModules();
 125     }
 126
 127     @Override
 128     public String getModuleSource(final ModuleIdentifier moduleIdentifier) {
 129         return getYangStoreSnapshot().getModuleSource(moduleIdentifier);
 130     }
 131
 132     public void refresh() {
 133         final YangStoreSnapshot previous = ref.get().get();
 134         ref.set(new SoftReference<YangStoreSnapshot>(null));
 135         notificationExecutor.submit(new CapabilityChangeNotifier(previous));
 136     }
 137
 138     public AutoCloseable registerCapabilityListener(final CapabilityListener listener) {
 139         if(ref.get() == null || ref.get().get() == null) {
 140             getYangStoreSnapshot();
 141         }
 142
 143         this.listeners.add(listener);
 144         listener.onCapabilitiesAdded(NetconfOperationServiceFactoryImpl.setupCapabilities(ref.get().get()));
 145
 146         return new AutoCloseable() {
 147             @Override
 148             public void close() throws Exception {
 149                 YangStoreService.this.listeners.remove(listener);
 150             }
 151         };
 152     }
 153
 154     private static final Function<Module, Capability> MODULE_TO_CAPABILITY = new Function<Module, Capability>() {
 155         @Override
 156         public Capability apply(final Module module) {
 157             return new NetconfOperationServiceFactoryImpl.YangStoreCapability(module, module.getSource());
 158         }
 159     };
 160
 161     private final class CapabilityChangeNotifier implements Runnable {
 162
 163         private final YangStoreSnapshot previous;
 164
 165         public CapabilityChangeNotifier(final YangStoreSnapshot previous) {
 166             this.previous = previous;
 167         }
 168
 169         @Override
 170         public void run() {
 171             final YangStoreContext current = getYangStoreSnapshot();
 172
 173             if(current.equals(previous) == false) {
 174                 final Sets.SetView<Module> removed = Sets.difference(previous.getModules(), current.getModules());
 175                 final Sets.SetView<Module> added = Sets.difference(current.getModules(), previous.getModules());
 176
 177                 // Notify notification manager
 178                 notificationPublisher.onCapabilityChanged(computeDiff(removed, added));
 179
 180                 // Notify direct capability listener TODO would it not be better if the capability listeners went through notification manager ?
 181                 for (final CapabilityListener listener : listeners) {
 182                     listener.onCapabilitiesAdded(Sets.newHashSet(Collections2.transform(added, MODULE_TO_CAPABILITY)));
 183                 }
 184                 for (final CapabilityListener listener : listeners) {
 185                     listener.onCapabilitiesRemoved(Sets.newHashSet(Collections2.transform(removed, MODULE_TO_CAPABILITY)));
 186                 }
 187             }
 188         }
 189     }
 190
 191     private static final Function<Module, Uri> MODULE_TO_URI = new Function<Module, Uri>() {
 192         @Override
 193         public Uri apply(final Module input) {
 194             return new Uri(new NetconfOperationServiceFactoryImpl.YangStoreCapability(input, input.getSource()).getCapabilityUri());
 195         }
 196     };
 197
 198     static NetconfCapabilityChange computeDiff(final Sets.SetView<Module> removed, final Sets.SetView<Module> added) {
 199         final NetconfCapabilityChangeBuilder netconfCapabilityChangeBuilder = new NetconfCapabilityChangeBuilder();
 200         netconfCapabilityChangeBuilder.setChangedBy(new ChangedByBuilder().setServerOrUser(new ServerBuilder().setServer(true).build()).build());
 201         netconfCapabilityChangeBuilder.setDeletedCapability(Lists.newArrayList(Collections2.transform(removed, MODULE_TO_URI)));
 202         netconfCapabilityChangeBuilder.setAddedCapability(Lists.newArrayList(Collections2.transform(added, MODULE_TO_URI)));
 203         // TODO modified should be computed ... but why ?
 204         netconfCapabilityChangeBuilder.setModifiedCapability(Collections.<Uri>emptyList());
 205         return netconfCapabilityChangeBuilder.build();
 206     }
 207
 208
 209     /**
 210      * Looks for NetconfNotificationCollector service and publishes base netconf notifications if possible
 211      */
 212     private static class NotificationCollectorTracker implements ServiceTrackerCustomizer<NetconfNotificationCollector, NetconfNotificationCollector>, BaseNetconfNotificationListener, AutoCloseable {
 213
 214         private final BundleContext context;
 215         private final ServiceTracker<NetconfNotificationCollector, NetconfNotificationCollector> listenerTracker;
 216         private BaseNotificationPublisherRegistration publisherReg;
 217
 218         public NotificationCollectorTracker(final BundleContext context) {
 219             this.context = context;
 220             listenerTracker = new ServiceTracker<>(context, NetconfNotificationCollector.class, this);
 221             listenerTracker.open();
 222         }
 223
 224         @Override
 225         public synchronized NetconfNotificationCollector addingService(final ServiceReference<NetconfNotificationCollector> reference) {
 226             closePublisherRegistration();
 227             publisherReg = context.getService(reference).registerBaseNotificationPublisher();
 228             return null;
 229         }
 230
 231         @Override
 232         public synchronized void modifiedService(final ServiceReference<NetconfNotificationCollector> reference, final NetconfNotificationCollector service) {
 233             closePublisherRegistration();
 234             publisherReg = context.getService(reference).registerBaseNotificationPublisher();
 235         }
 236
 237         @Override
 238         public synchronized void removedService(final ServiceReference<NetconfNotificationCollector> reference, final NetconfNotificationCollector service) {
 239             closePublisherRegistration();
 240             publisherReg = null;
 241         }
 242
 243         private void closePublisherRegistration() {
 244             if(publisherReg != null) {
 245                 publisherReg.close();
 246             }
 247         }
 248
 249         @Override
 250         public synchronized void close() {
 251             closePublisherRegistration();
 252             listenerTracker.close();
 253         }
 254
 255         @Override
 256         public void onCapabilityChanged(final NetconfCapabilityChange capabilityChange) {
 257             if(publisherReg == null) {
 258                 LOG.warn("Omitting notification due to missing notification service: {}", capabilityChange);
 259                 return;
 260             }
 261
 262             publisherReg.onCapabilityChanged(capabilityChange);
 263         }
 264     }
 265 }