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