Minor changes to config-persister to allow for karaf

These minor changes expose a ConfigPusher service for use
by a config-persister-feature-adapter (to follow) that
can push configs attached to features.

Change-Id: I382d2ab578d19ea9c24afa521a9a5a047360fc50
Signed-off-by: Ed Warnicke <eaw@cisco.com>

diff --git a/opendaylight/config/config-persister-api/src/main/java/org/opendaylight/controller/config/persist/api/ConfigPusher.java b/opendaylight/config/config-persister-api/src/main/java/org/opendaylight/controller/config/persist/api/ConfigPusher.java
new file mode 100644 (file)
index 0000000..2dade8a
--- /dev/null
+++ b/opendaylight/config/config-persister-api/src/main/java/org/opendaylight/controller/config/persist/api/ConfigPusher.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright (c) 2013 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.controller.config.persist.api;
+
+import java.util.List;
+/*
+ * The config pusher service pushes configs into the config subsystem
+ */
+public interface ConfigPusher {
+
+    /*
+     * Pushes configs into the config subsystem
+     */
+
+    public void pushConfigs(List<? extends ConfigSnapshotHolder> configs) throws InterruptedException;
+}

diff --git a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusher.java b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusherImpl.java
similarity index 85%
rename from opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusher.java
rename to opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusherImpl.java
index fff8d611b7975b34dce89ce603f164603ea2b582..5f311b5232ed676263925ce6af94b779c662df2e 100644 (file)
--- a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusher.java
+++ b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusherImpl.java
@@ -10,9 +10,6 @@ package org.opendaylight.controller.netconf.persist.impl;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
 
 import static com.google.common.base.Preconditions.checkNotNull;
 

-import com.google.common.base.Function;
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.Collections2;

 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collection;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collection;


@@ -23,10 +20,17 @@ import java.util.Map.Entry;
 import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeMap;
 import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeMap;

+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;

+

 import javax.annotation.concurrent.Immutable;
 import javax.annotation.concurrent.Immutable;

+import javax.management.MBeanServerConnection;
+

 import org.opendaylight.controller.config.api.ConflictingVersionException;
 import org.opendaylight.controller.config.api.ConflictingVersionException;

+import org.opendaylight.controller.config.persist.api.ConfigPusher;

 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;

+import org.opendaylight.controller.config.persist.api.Persister;

 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
 import org.opendaylight.controller.netconf.api.NetconfMessage;
 import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
 import org.opendaylight.controller.netconf.api.NetconfMessage;
 import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;


@@ -45,22 +49,60 @@ import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.xml.sax.SAXException;
 
 import org.w3c.dom.Element;
 import org.xml.sax.SAXException;
 

+import com.google.common.base.Function;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Collections2;
+

 @Immutable
 @Immutable

-public class ConfigPusher {
-    private static final Logger logger = LoggerFactory.getLogger(ConfigPusher.class);
+public class ConfigPusherImpl implements ConfigPusher {
+    private static final Logger logger = LoggerFactory.getLogger(ConfigPusherImpl.class);

 
     private final long maxWaitForCapabilitiesMillis;
     private final long conflictingVersionTimeoutMillis;
     private final NetconfOperationServiceFactory configNetconfConnector;
 
     private final long maxWaitForCapabilitiesMillis;
     private final long conflictingVersionTimeoutMillis;
     private final NetconfOperationServiceFactory configNetconfConnector;

+    private static final int QUEUE_SIZE = 100;
+    private BlockingQueue<List<? extends ConfigSnapshotHolder>> queue = new LinkedBlockingQueue<List<? extends ConfigSnapshotHolder>>(QUEUE_SIZE);

 
 

-    public ConfigPusher(NetconfOperationServiceFactory configNetconfConnector, long maxWaitForCapabilitiesMillis,
+    public ConfigPusherImpl(NetconfOperationServiceFactory configNetconfConnector, long maxWaitForCapabilitiesMillis,

                         long conflictingVersionTimeoutMillis) {
         this.configNetconfConnector = configNetconfConnector;
         this.maxWaitForCapabilitiesMillis = maxWaitForCapabilitiesMillis;
         this.conflictingVersionTimeoutMillis = conflictingVersionTimeoutMillis;
     }
 
                         long conflictingVersionTimeoutMillis) {
         this.configNetconfConnector = configNetconfConnector;
         this.maxWaitForCapabilitiesMillis = maxWaitForCapabilitiesMillis;
         this.conflictingVersionTimeoutMillis = conflictingVersionTimeoutMillis;
     }
 

-    public synchronized LinkedHashMap<ConfigSnapshotHolder, EditAndCommitResponse> pushConfigs(List<ConfigSnapshotHolder> configs) throws NetconfDocumentedException {
+    public void process(List<AutoCloseable> autoCloseables, MBeanServerConnection platformMBeanServer, Persister persisterAggregator) throws InterruptedException {
+        List<? extends ConfigSnapshotHolder> configs;
+        while(true) {
+            configs = queue.take();
+            try {
+                internalPushConfigs(configs);
+                ConfigPersisterNotificationHandler jmxNotificationHandler = new ConfigPersisterNotificationHandler(platformMBeanServer, persisterAggregator);
+                synchronized (autoCloseables) {
+                    autoCloseables.add(jmxNotificationHandler);
+                }
+                /*
+                 * We have completed initial configuration. At this point
+                 * it is good idea to perform garbage collection to prune
+                 * any garbage we have accumulated during startup.
+                 */
+                logger.debug("Running post-initialization garbage collection...");
+                System.gc();
+                logger.debug("Post-initialization garbage collection completed.");
+                logger.debug("ConfigPusher has pushed configs {}, gc completed", configs);
+            }
+            catch (NetconfDocumentedException e) {
+                logger.error("Error pushing configs {}",configs);
+                throw new IllegalStateException(e);
+            }
+        }
+    }
+
+    public void pushConfigs(List<? extends ConfigSnapshotHolder> configs) throws InterruptedException {
+        logger.debug("Requested to push configs {}", configs);
+        this.queue.put(configs);
+    }
+
+    private LinkedHashMap<? extends ConfigSnapshotHolder, EditAndCommitResponse> internalPushConfigs(List<? extends ConfigSnapshotHolder> configs) throws NetconfDocumentedException {

         logger.debug("Last config snapshots to be pushed to netconf: {}", configs);
         LinkedHashMap<ConfigSnapshotHolder, EditAndCommitResponse> result = new LinkedHashMap<>();
         // start pushing snapshots:
         logger.debug("Last config snapshots to be pushed to netconf: {}", configs);
         LinkedHashMap<ConfigSnapshotHolder, EditAndCommitResponse> result = new LinkedHashMap<>();
         // start pushing snapshots:


@@ -278,7 +320,7 @@ public class ConfigPusher {
 
     private static NetconfMessage getCommitMessage() {
         String resource = "/netconfOp/commit.xml";
 
     private static NetconfMessage getCommitMessage() {
         String resource = "/netconfOp/commit.xml";

-        try (InputStream stream = ConfigPusher.class.getResourceAsStream(resource)) {
+        try (InputStream stream = ConfigPusherImpl.class.getResourceAsStream(resource)) {

             checkNotNull(stream, "Unable to load resource " + resource);
             return new NetconfMessage(XmlUtil.readXmlToDocument(stream));
         } catch (SAXException | IOException e) {
             checkNotNull(stream, "Unable to load resource " + resource);
             return new NetconfMessage(XmlUtil.readXmlToDocument(stream));
         } catch (SAXException | IOException e) {

diff --git a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java
index 48ae0cb91a132d1cd7bff773df604e2d5ae1c47a..0a48e6c67dc8cc290c2b81a1b0eb67c05ac7c618 100644 (file)
--- a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java
+++ b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java
@@ -8,13 +8,18 @@
 
 package org.opendaylight.controller.netconf.persist.impl.osgi;
 
 
 package org.opendaylight.controller.netconf.persist.impl.osgi;
 

-import com.google.common.annotations.VisibleForTesting;
+import java.lang.management.ManagementFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import javax.management.MBeanServer;
+
+import org.opendaylight.controller.config.persist.api.ConfigPusher;

 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;

-import org.opendaylight.controller.netconf.api.NetconfDocumentedException;

 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;

-import org.opendaylight.controller.netconf.persist.impl.ConfigPersisterNotificationHandler;
-import org.opendaylight.controller.netconf.persist.impl.ConfigPusher;
+import org.opendaylight.controller.netconf.persist.impl.ConfigPusherImpl;

 import org.opendaylight.controller.netconf.persist.impl.PersisterAggregator;
 import org.opendaylight.controller.netconf.util.CloseableUtil;
 import org.osgi.framework.BundleActivator;
 import org.opendaylight.controller.netconf.persist.impl.PersisterAggregator;
 import org.opendaylight.controller.netconf.util.CloseableUtil;
 import org.osgi.framework.BundleActivator;


@@ -23,16 +28,13 @@ import org.osgi.framework.Constants;
 import org.osgi.framework.Filter;
 import org.osgi.framework.InvalidSyntaxException;
 import org.osgi.framework.ServiceReference;
 import org.osgi.framework.Filter;
 import org.osgi.framework.InvalidSyntaxException;
 import org.osgi.framework.ServiceReference;

+import org.osgi.framework.ServiceRegistration;

 import org.osgi.util.tracker.ServiceTracker;
 import org.osgi.util.tracker.ServiceTrackerCustomizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.osgi.util.tracker.ServiceTracker;
 import org.osgi.util.tracker.ServiceTrackerCustomizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

-import javax.management.MBeanServer;
-import java.lang.management.ManagementFactory;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
+import com.google.common.annotations.VisibleForTesting;

 
 public class ConfigPersisterActivator implements BundleActivator {
 
 
 public class ConfigPersisterActivator implements BundleActivator {
 


@@ -49,11 +51,15 @@ public class ConfigPersisterActivator implements BundleActivator {
     public static final String STORAGE_ADAPTER_CLASS_PROP_SUFFIX = "storageAdapterClass";
 
     private List<AutoCloseable> autoCloseables;
     public static final String STORAGE_ADAPTER_CLASS_PROP_SUFFIX = "storageAdapterClass";
 
     private List<AutoCloseable> autoCloseables;

+    private volatile BundleContext context;

 
 

+    ServiceRegistration<?> registration;

 
     @Override
     public void start(final BundleContext context) throws Exception {
         logger.debug("ConfigPersister starting");
 
     @Override
     public void start(final BundleContext context) throws Exception {
         logger.debug("ConfigPersister starting");

+        this.context = context;
+

         autoCloseables = new ArrayList<>();
         PropertiesProviderBaseImpl propertiesProvider = new PropertiesProviderBaseImpl(context);
 
         autoCloseables = new ArrayList<>();
         PropertiesProviderBaseImpl propertiesProvider = new PropertiesProviderBaseImpl(context);
 


@@ -81,8 +87,14 @@ public class ConfigPersisterActivator implements BundleActivator {
     }
 
     @Override
     }
 
     @Override

-    public synchronized void stop(BundleContext context) throws Exception {
-        CloseableUtil.closeAll(autoCloseables);
+    public void stop(BundleContext context) throws Exception {
+        synchronized(autoCloseables) {
+            CloseableUtil.closeAll(autoCloseables);
+            if (registration != null) {
+                registration.unregister();
+            }
+            this.context = null;
+        }

     }
 
 
     }
 
 


@@ -147,35 +159,29 @@ public class ConfigPersisterActivator implements BundleActivator {
             logger.trace("Got InnerCustomizer.addingService {}", reference);
             NetconfOperationServiceFactory service = reference.getBundle().getBundleContext().getService(reference);
 
             logger.trace("Got InnerCustomizer.addingService {}", reference);
             NetconfOperationServiceFactory service = reference.getBundle().getBundleContext().getService(reference);
 

-            final ConfigPusher configPusher = new ConfigPusher(service, maxWaitForCapabilitiesMillis, conflictingVersionTimeoutMillis);
+            logger.debug("Creating new job queue");
+
+            final ConfigPusherImpl configPusher = new ConfigPusherImpl(service, maxWaitForCapabilitiesMillis, conflictingVersionTimeoutMillis);

             logger.debug("Configuration Persister got {}", service);
             logger.debug("Configuration Persister got {}", service);

+            logger.debug("Context was {}", context);
+            logger.debug("Registration was {}", registration);
+

             final Thread pushingThread = new Thread(new Runnable() {
                 @Override
                 public void run() {
                     try {
             final Thread pushingThread = new Thread(new Runnable() {
                 @Override
                 public void run() {
                     try {

-                        configPusher.pushConfigs(configs);
-                    } catch (NetconfDocumentedException e) {
-                        logger.error("Error pushing configs {}",configs);
-                        throw new IllegalStateException(e);
+                        if(configs != null && !configs.isEmpty()) {
+                            configPusher.pushConfigs(configs);
+                        }
+                        registration = context.registerService(ConfigPusher.class.getName(), configPusher, null);
+                        configPusher.process(autoCloseables, platformMBeanServer, persisterAggregator);
+                    } catch (InterruptedException e) {
+                        logger.info("ConfigPusher thread stopped",e);

                     }
                     logger.info("Configuration Persister initialization completed.");
                     }
                     logger.info("Configuration Persister initialization completed.");

-
-                    /*
-                     * We have completed initial configuration. At this point
-                     * it is good idea to perform garbage collection to prune
-                     * any garbage we have accumulated during startup.
-                     */
-                    logger.debug("Running post-initialization garbage collection...");
-                    System.gc();
-                    logger.debug("Post-initialization garbage collection completed.");
-
-                    ConfigPersisterNotificationHandler jmxNotificationHandler = new ConfigPersisterNotificationHandler(platformMBeanServer, persisterAggregator);
-                    synchronized (ConfigPersisterActivator.this) {
-                        autoCloseables.add(jmxNotificationHandler);
-                    }

                 }
             }, "config-pusher");
                 }
             }, "config-pusher");

-            synchronized (ConfigPersisterActivator.this) {
+            synchronized (autoCloseables) {

                 autoCloseables.add(new AutoCloseable() {
                     @Override
                     public void close() {
                 autoCloseables.add(new AutoCloseable() {
                     @Override
                     public void close() {

diff --git a/opendaylight/netconf/config-persister-impl/src/test/java/org/opendaylight/controller/netconf/persist/impl/osgi/MockedBundleContext.java b/opendaylight/netconf/config-persister-impl/src/test/java/org/opendaylight/controller/netconf/persist/impl/osgi/MockedBundleContext.java
index 95fd5f65498e507c54e6d88023dc24b25475031a..3e5249468da703051ed8fe436843d37c316ab7c5 100644 (file)
--- a/opendaylight/netconf/config-persister-impl/src/test/java/org/opendaylight/controller/netconf/persist/impl/osgi/MockedBundleContext.java
+++ b/opendaylight/netconf/config-persister-impl/src/test/java/org/opendaylight/controller/netconf/persist/impl/osgi/MockedBundleContext.java
@@ -7,10 +7,23 @@
  */
 package org.opendaylight.controller.netconf.persist.impl.osgi;
 
  */
 package org.opendaylight.controller.netconf.persist.impl.osgi;
 

-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Dictionary;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+

 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;

+import org.opendaylight.controller.config.persist.api.ConfigPusher;

 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
 import org.opendaylight.controller.config.persist.api.Persister;
 import org.opendaylight.controller.config.persist.api.PropertiesProvider;
 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
 import org.opendaylight.controller.config.persist.api.Persister;
 import org.opendaylight.controller.config.persist.api.PropertiesProvider;


@@ -23,18 +36,10 @@ import org.osgi.framework.BundleContext;
 import org.osgi.framework.Filter;
 import org.osgi.framework.ServiceListener;
 import org.osgi.framework.ServiceReference;
 import org.osgi.framework.Filter;
 import org.osgi.framework.ServiceListener;
 import org.osgi.framework.ServiceReference;

+import org.osgi.framework.ServiceRegistration;

 
 

-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.SortedSet;
-import java.util.TreeSet;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doReturn;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;

 
 final class MockedBundleContext {
     @Mock
 
 final class MockedBundleContext {
     @Mock


@@ -49,6 +54,8 @@ final class MockedBundleContext {
     NetconfOperationServiceFactory serviceFactory;
     @Mock
     private NetconfOperationService service;
     NetconfOperationServiceFactory serviceFactory;
     @Mock
     private NetconfOperationService service;

+    @Mock
+    private ServiceRegistration<?> registration;

 
     MockedBundleContext(long maxWaitForCapabilitiesMillis, long conflictingVersionTimeoutMillis) throws Exception {
         MockitoAnnotations.initMocks(this);
 
     MockedBundleContext(long maxWaitForCapabilitiesMillis, long conflictingVersionTimeoutMillis) throws Exception {
         MockitoAnnotations.initMocks(this);


@@ -77,6 +84,11 @@ final class MockedBundleContext {
         doReturn(Collections.emptySet()).when(service).getCapabilities();
         doNothing().when(service).close();
         doReturn("serviceFactoryMock").when(serviceFactory).toString();
         doReturn(Collections.emptySet()).when(service).getCapabilities();
         doNothing().when(service).close();
         doReturn("serviceFactoryMock").when(serviceFactory).toString();

+
+        doNothing().when(registration).unregister();
+        doReturn(registration).when(context).registerService(
+                eq(ConfigPusher.class.getName()), any(Closeable.class),
+                any(Dictionary.class));

     }
 
     public BundleContext getBundleContext() {
     }
 
     public BundleContext getBundleContext() {