cxf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dk...@apache.org
Subject svn commit: r1179822 - in /cxf/trunk: parent/ rt/core/ rt/core/src/main/java/org/apache/cxf/bus/osgi/ rt/core/src/main/java/org/apache/cxf/workqueue/
Date Thu, 06 Oct 2011 20:23:03 GMT
Author: dkulp
Date: Thu Oct  6 20:23:03 2011
New Revision: 1179822

URL: http://svn.apache.org/viewvc?rev=1179822&view=rev
Log:
Add ability to configure global workqueues in OSGi

Added:
    cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java  
(with props)
Modified:
    cxf/trunk/parent/pom.xml
    cxf/trunk/rt/core/pom.xml
    cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiExtensionLocator.java
    cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java

Modified: cxf/trunk/parent/pom.xml
URL: http://svn.apache.org/viewvc/cxf/trunk/parent/pom.xml?rev=1179822&r1=1179821&r2=1179822&view=diff
==============================================================================
--- cxf/trunk/parent/pom.xml (original)
+++ cxf/trunk/parent/pom.xml Thu Oct  6 20:23:03 2011
@@ -438,6 +438,12 @@
                 <scope>provided</scope>
             </dependency>
             <dependency>
+                <groupId>org.osgi</groupId>
+                <artifactId>org.osgi.compendium</artifactId>
+                <version>${cxf.osgi.version}</version>
+                <scope>provided</scope>
+            </dependency>
+            <dependency>
                 <groupId>cglib</groupId>
                 <artifactId>cglib-nodep</artifactId>
                 <version>2.2.2</version>

Modified: cxf/trunk/rt/core/pom.xml
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/core/pom.xml?rev=1179822&r1=1179821&r2=1179822&view=diff
==============================================================================
--- cxf/trunk/rt/core/pom.xml (original)
+++ cxf/trunk/rt/core/pom.xml Thu Oct  6 20:23:03 2011
@@ -41,6 +41,12 @@
         <dependency>
             <groupId>org.osgi</groupId>
             <artifactId>org.osgi.core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.osgi</groupId>
+            <artifactId>org.osgi.compendium</artifactId>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.cxf</groupId>

Added: cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java?rev=1179822&view=auto
==============================================================================
--- cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java (added)
+++ cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java Thu
Oct  6 20:23:03 2011
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cxf.bus.osgi;
+
+import java.util.Dictionary;
+
+import org.apache.cxf.workqueue.AutomaticWorkQueueImpl;
+import org.osgi.service.cm.ConfigurationException;
+import org.osgi.service.cm.ManagedService;
+
+/**
+ * 
+ */
+public class OSGiAutomaticWorkQueue extends AutomaticWorkQueueImpl implements ManagedService
{
+    
+    public OSGiAutomaticWorkQueue(String name) {
+        super(name);
+    }
+
+    /** {@inheritDoc}*/
+    public void updated(Dictionary d) throws ConfigurationException {
+        String name = getName();
+        String s = (String)d.get("org.apache.cxf.workqueue." + name + ".highWaterMark");
+        if (s != null) {
+            setHighWaterMark(Integer.parseInt(s));
+        }
+        s = (String)d.get("org.apache.cxf.workqueue." + name + ".lowWaterMark");
+        if (s != null) {
+            setLowWaterMark(Integer.parseInt(s));
+        }
+        s = (String)d.get("org.apache.cxf.workqueue." + name + ".initialSize");
+        if (s != null) {
+            setInitialSize(Integer.parseInt(s));
+        }
+        s = (String)d.get("org.apache.cxf.workqueue." + name + ".dequeueTimeout");
+        if (s != null) {
+            setDequeueTimeout(Long.parseLong(s));
+        }
+        s = (String)d.get("org.apache.cxf.workqueue." + name + ".queueSize");
+        if (s != null) {
+            setQueueSize(Integer.parseInt(s));
+        } 
+
+    }
+
+}

Propchange: cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiAutomaticWorkQueue.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Modified: cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiExtensionLocator.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiExtensionLocator.java?rev=1179822&r1=1179821&r2=1179822&view=diff
==============================================================================
--- cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiExtensionLocator.java (original)
+++ cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/osgi/OSGiExtensionLocator.java Thu
Oct  6 20:23:03 2011
@@ -39,14 +39,20 @@ import org.apache.cxf.bus.extension.Exte
 import org.apache.cxf.buslifecycle.BusLifeCycleListener;
 import org.apache.cxf.buslifecycle.BusLifeCycleManager;
 import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.workqueue.AutomaticWorkQueueImpl;
+import org.apache.cxf.workqueue.WorkQueueManager;
 import org.osgi.framework.Bundle;
 import org.osgi.framework.BundleActivator;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.BundleEvent;
 import org.osgi.framework.Constants;
+import org.osgi.framework.ServiceReference;
 import org.osgi.framework.ServiceRegistration;
 import org.osgi.framework.SynchronousBundleListener;
 import org.osgi.framework.Version;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
+import org.osgi.service.cm.ManagedService;
 
 /**
  * 
@@ -58,6 +64,11 @@ public class OSGiExtensionLocator implem
     private long id;
     private Extension listener;
 
+    static class WorkQueueList {
+        List<AutomaticWorkQueueImpl> list = new CopyOnWriteArrayList<AutomaticWorkQueueImpl>();
+    };
+    private WorkQueueList workQueues = new WorkQueueList();
+    
     /** {@inheritDoc}*/
     public void bundleChanged(BundleEvent event) {
         if (event.getType() == BundleEvent.RESOLVED && id != event.getBundle().getBundleId())
{
@@ -85,6 +96,43 @@ public class OSGiExtensionLocator implem
                 register(bundle);
             }
         }
+        ServiceReference configAdminServiceRef =  
+            context.getServiceReference(ConfigurationAdmin.class.getName());  
+              
+        if (configAdminServiceRef != null) {  
+            ConfigurationAdmin configAdmin = (ConfigurationAdmin)  
+                    context.getService(configAdminServiceRef);  
+              
+            Configuration config = configAdmin.getConfiguration("org.apache.cxf.workqueues");
+            Dictionary d = config.getProperties();
+            if (d != null) {
+                Properties props = new Properties();
+                props.put(Constants.SERVICE_PID, "org.apache.cxf.workqueues");  
+                String s = (String)d.get("org.apache.cxf.workqueue.names");
+                String s2[] = s.split(",");
+                for (String name : s2) {
+                    name = name.trim();
+                    OSGiAutomaticWorkQueue wq = new OSGiAutomaticWorkQueue(name);
+                    wq.updated(d);
+                    wq.setShared(true);
+                    workQueues.list.add(wq);
+                    
+                    context.registerService(ManagedService.class.getName(),  
+                                            wq, props); 
+                }
+            }
+            if (!workQueues.list.isEmpty()) {
+                Extension ext = new Extension(WorkQueueList.class) {
+                    public Object getLoadedObject() {
+                        return workQueues;
+                    }
+                    public Extension cloneNoObject() {
+                        return this;
+                    }
+                };
+                ExtensionRegistry.addExtensions(Collections.singletonList(ext));
+            }
+        }
     }
 
     /** {@inheritDoc}*/
@@ -94,6 +142,11 @@ public class OSGiExtensionLocator implem
         while (!extensions.isEmpty()) {
             unregister(extensions.keySet().iterator().next());
         }
+        for (AutomaticWorkQueueImpl wq : workQueues.list) {
+            wq.setShared(false);
+            wq.shutdown(true);
+        }
+        workQueues.list.clear();
     }
     private void registerBusListener() {
         listener = new Extension(OSGIBusListener.class);
@@ -155,6 +208,16 @@ public class OSGiExtensionLocator implem
         }
  
         public void initComplete() {
+            WorkQueueManager m = bus.getExtension(WorkQueueManager.class);
+            WorkQueueList l = bus.getExtension(WorkQueueList.class);
+            if (l != null && m != null) {
+                for (AutomaticWorkQueueImpl wq : l.list) {
+                    if (m.getNamedWorkQueue(wq.getName()) == null) {
+                        m.addNamedWorkQueue(wq.getName(), wq);
+                    }
+                }
+            }
+            
             BundleContext context = bus.getExtension(BundleContext.class);
             if (context != null) {
                 Properties props = new Properties();

Modified: cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java?rev=1179822&r1=1179821&r2=1179822&view=diff
==============================================================================
--- cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java (original)
+++ cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java Thu
Oct  6 20:23:03 2011
@@ -65,6 +65,8 @@ public class AutomaticWorkQueueImpl impl
     DelayQueue<DelayedTaskWrapper> delayQueue;
     WatchDog watchDog;
     
+    boolean shared;
+    
     public AutomaticWorkQueueImpl() {
         this(DEFAULT_MAX_QUEUE_SIZE);
     }    
@@ -103,6 +105,10 @@ public class AutomaticWorkQueueImpl impl
         this.name = name;
     }
     
+    public void setShared(boolean b) {
+        shared = b;
+    }
+    
     protected synchronized ThreadPoolExecutor getExecutor() {
         if (executor == null) {
             threadFactory = createThreadFactory(name);
@@ -412,7 +418,7 @@ public class AutomaticWorkQueueImpl impl
     // AutomaticWorkQueue interface
     
     public void shutdown(boolean processRemainingWorkItems) {
-        if (executor != null) {
+        if (executor != null && !shared) {
             if (!processRemainingWorkItems) {
                 executor.getQueue().clear();
             }



Mime
View raw message