Author: dkulp
Date: Thu Oct 6 16:15:26 2011
New Revision: 1179688
URL: http://svn.apache.org/viewvc?rev=1179688&view=rev
Log:
Updates to add a <cxf:workqueue> element to the core schema to make it
easier to configure the workqueue parameters.
Delay the creation of the actual threadpool till first call to execute
to allow configuration ahead of time.
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/configuration/ConfiguredBeanLocator.java
cxf/trunk/api/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueue.java
cxf/trunk/common/common/src/main/java/org/apache/cxf/common/util/ReflectionUtil.java
cxf/trunk/common/common/src/main/java/org/apache/cxf/configuration/spring/AbstractBeanDefinitionParser.java
cxf/trunk/common/common/src/main/resources/META-INF/spring.schemas
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/CXFBusImpl.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/BlueprintBeanLocator.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/CXFCoreNamespaceHandler.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/extension/ExtensionManagerImpl.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/NamespaceHandler.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/SpringBeanLocator.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/configuration/blueprint/AbstractBPBeanDefinitionParser.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueueImpl.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueImplMBeanWrapper.java
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueManagerImpl.java
cxf/trunk/rt/core/src/main/resources/schemas/blueprint/core.xsd
cxf/trunk/rt/core/src/main/resources/schemas/core.xsd
cxf/trunk/rt/core/src/test/java/org/apache/cxf/workqueue/AutomaticWorkQueueTest.java
cxf/trunk/rt/management/src/test/java/org/apache/cxf/management/InstrumentationManagerTest.java
cxf/trunk/rt/management/src/test/resources/managed-spring.xml
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/configuration/ConfiguredBeanLocator.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/configuration/ConfiguredBeanLocator.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/configuration/ConfiguredBeanLocator.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/configuration/ConfiguredBeanLocator.java
Thu Oct 6 16:15:26 2011
@@ -35,6 +35,15 @@ public interface ConfiguredBeanLocator {
*/
List<String> getBeanNamesOfType(Class<?> type);
+
+ /**
+ * Gets the bean of the given name and type
+ * @param name
+ * @param type
+ * @return the bean
+ */
+ <T> T getBeanOfType(String name, Class<T> type);
+
/**
* Gets all the configured beans of the specific types. Causes them
* all to be loaded.
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueue.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueue.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueue.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/workqueue/AutomaticWorkQueue.java
Thu Oct 6 16:15:26 2011
@@ -20,6 +20,12 @@ package org.apache.cxf.workqueue;
public interface AutomaticWorkQueue extends WorkQueue {
/**
+ * Get's the name of the workqueue
+ * @return the name
+ */
+ String getName();
+
+ /**
* Initiates an orderly shutdown.
* If <code>processRemainingWorkItems</code>
* is true, waits for all active items to finish execution before
returning, otherwise returns
Modified:
cxf/trunk/common/common/src/main/java/org/apache/cxf/common/util/ReflectionUtil.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/common/common/src/main/java/org/apache/cxf/common/util/ReflectionUtil.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/common/common/src/main/java/org/apache/cxf/common/util/ReflectionUtil.java
(original)
+++
cxf/trunk/common/common/src/main/java/org/apache/cxf/common/util/ReflectionUtil.java
Thu Oct 6 16:15:26 2011
@@ -102,19 +102,19 @@ public final class ReflectionUtil {
});
}
- public static void setAccessible(final AccessibleObject o) {
- AccessController.doPrivileged(new PrivilegedAction<Boolean>() {
- public Boolean run() {
+ public static <T extends AccessibleObject> T setAccessible(final T o) {
+ return AccessController.doPrivileged(new PrivilegedAction<T>() {
+ public T run() {
o.setAccessible(true);
- return true;
+ return o;
}
});
}
- public static void setAccessible(final AccessibleObject o, final boolean
b) {
- AccessController.doPrivileged(new PrivilegedAction<Boolean>() {
- public Boolean run() {
+ public static <T extends AccessibleObject> T setAccessible(final T o,
final boolean b) {
+ return AccessController.doPrivileged(new PrivilegedAction<T>() {
+ public T run() {
o.setAccessible(b);
- return true;
+ return o;
}
});
}
Modified:
cxf/trunk/common/common/src/main/java/org/apache/cxf/configuration/spring/AbstractBeanDefinitionParser.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/common/common/src/main/java/org/apache/cxf/configuration/spring/AbstractBeanDefinitionParser.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/common/common/src/main/java/org/apache/cxf/configuration/spring/AbstractBeanDefinitionParser.java
(original)
+++
cxf/trunk/common/common/src/main/java/org/apache/cxf/configuration/spring/AbstractBeanDefinitionParser.java
Thu Oct 6 16:15:26 2011
@@ -63,6 +63,9 @@ public abstract class AbstractBeanDefini
private JAXBContext context;
private Set<Class<?>> classes;
+ public AbstractBeanDefinitionParser() {
+ }
+
@Override
protected void doParse(Element element, ParserContext ctx,
BeanDefinitionBuilder bean) {
boolean setBus = parseAttributes(element, ctx, bean);
@@ -93,7 +96,9 @@ public abstract class AbstractBeanDefini
bean.setAbstract(true);
} else if ("depends-on".equals(name)) {
bean.addDependsOn(val);
- } else if (!"id".equals(name) && !"name".equals(name) &&
isAttribute(pre, name)) {
+ } else if ("name".equals(name)) {
+ processNameAttribute(element, ctx, bean, val);
+ } else if (!"id".equals(name) && isAttribute(pre, name)) {
if ("bus".equals(name)) {
if (val != null && val.trim().length() > 0
&& ctx.getRegistry().containsBeanDefinition(val)) {
@@ -108,6 +113,13 @@ public abstract class AbstractBeanDefini
return setBus;
}
+ protected void processNameAttribute(Element element,
+ ParserContext ctx,
+ BeanDefinitionBuilder bean,
+ String val) {
+ //nothing
+ }
+
private boolean isNamespace(String name, String prefix) {
return "xmlns".equals(prefix) || prefix == null &&
"xmlns".equals(name);
}
Modified: cxf/trunk/common/common/src/main/resources/META-INF/spring.schemas
URL:
http://svn.apache.org/viewvc/cxf/trunk/common/common/src/main/resources/META-INF/spring.schemas?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
--- cxf/trunk/common/common/src/main/resources/META-INF/spring.schemas
(original)
+++ cxf/trunk/common/common/src/main/resources/META-INF/spring.schemas Thu Oct
6 16:15:26 2011
@@ -18,4 +18,5 @@
# under the License.
#
#
-http\://cxf.apache.org/schemas/configuration/cxf-beans.xsd=schemas/configuration/cxf-beans.xsd
\ No newline at end of file
+http\://cxf.apache.org/schemas/configuration/cxf-beans.xsd=schemas/configuration/cxf-beans.xsd
+http\://cxf.apache.org/configuration/parameterized-types=schemas/configuration/parameterized-types.xsd
\ No newline at end of file
Modified: cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/CXFBusImpl.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/CXFBusImpl.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
--- cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/CXFBusImpl.java
(original)
+++ cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/CXFBusImpl.java Thu Oct
6 16:15:26 2011
@@ -129,6 +129,9 @@ public class CXFBusImpl extends Abstract
String value) {
return false;
}
+ public <T> T getBeanOfType(String name, Class<T> type) {
+ return null;
+ }
};
this.setExtension(loc, ConfiguredBeanLocator.class);
}
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/BlueprintBeanLocator.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/BlueprintBeanLocator.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/BlueprintBeanLocator.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/BlueprintBeanLocator.java
Thu Oct 6 16:15:26 2011
@@ -59,7 +59,20 @@ public class BlueprintBeanLocator implem
}
}
-
+ public <T> T getBeanOfType(String name, Class<T> type) {
+ ExecutionContext origContext
+ =
ExecutionContext.Holder.setContext((ExecutionContext)container.getRepository());
+ try {
+ Recipe r = container.getRepository().getRecipe(name);
+ if (r instanceof BeanRecipe && ((BeanRecipe)r).getType() != null
+ && type.isAssignableFrom(((BeanRecipe)r).getType())) {
+ return type.cast(container.getComponentInstance(name));
+ }
+ } finally {
+ ExecutionContext.Holder.setContext(origContext);
+ }
+ return orig.getBeanOfType(name, type);
+ }
/** {@inheritDoc}*/
public List<String> getBeanNamesOfType(Class<?> type) {
Set<String> names = new LinkedHashSet<String>();
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/CXFCoreNamespaceHandler.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/CXFCoreNamespaceHandler.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/CXFCoreNamespaceHandler.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/blueprint/CXFCoreNamespaceHandler.java
Thu Oct 6 16:15:26 2011
@@ -27,9 +27,11 @@ import org.w3c.dom.Node;
import org.apache.aries.blueprint.NamespaceHandler;
import org.apache.aries.blueprint.ParserContext;
+import org.apache.aries.blueprint.mutable.MutableBeanMetadata;
import org.apache.cxf.configuration.blueprint.SimpleBPBeanDefinitionParser;
import org.apache.cxf.feature.FastInfosetFeature;
import org.apache.cxf.feature.LoggingFeature;
+import org.apache.cxf.workqueue.AutomaticWorkQueueImpl;
import org.osgi.service.blueprint.container.BlueprintContainer;
import org.osgi.service.blueprint.reflect.ComponentMetadata;
import org.osgi.service.blueprint.reflect.Metadata;
@@ -57,6 +59,23 @@ public class CXFCoreNamespaceHandler imp
} else if ("fastinfoset".equals(s)) {
//fastinfosetfeature
return new
SimpleBPBeanDefinitionParser(FastInfosetFeature.class).parse(element, context);
+ } else if ("workqueue".equals(s)) {
+ //fastinfosetfeature
+ return new
SimpleBPBeanDefinitionParser(AutomaticWorkQueueImpl.class) {
+ public String getId(Element element, ParserContext context) {
+ String id = element.hasAttribute("id") ?
element.getAttribute("id") : null;
+ if (id == null) {
+ id = "cxf.workqueue.";
+ id += element.hasAttribute("name") ?
element.getAttribute("name") : "def";
+ }
+ return super.getId(element, context);
+ }
+
+ protected void processNameAttribute(Element element,
ParserContext ctx,
+ MutableBeanMetadata bean,
String val) {
+ bean.addProperty("name", createValue(ctx, val));
+ }
+ } .parse(element, context);
}
return null;
}
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/extension/ExtensionManagerImpl.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/extension/ExtensionManagerImpl.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/extension/ExtensionManagerImpl.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/extension/ExtensionManagerImpl.java
Thu Oct 6 16:15:26 2011
@@ -262,6 +262,16 @@ public class ExtensionManagerImpl implem
}
return ret;
}
+ public synchronized <T> T getBeanOfType(String name, Class<T> type) {
+ Extension ex = all.get(name);
+ if (ex != null) {
+ if (ex.getLoadedObject() == null) {
+ loadAndRegister(ex);
+ }
+ return type.cast(ex.getLoadedObject());
+ }
+ return null;
+ }
public synchronized <T> Collection<? extends T> getBeansOfType(Class<T>
type) {
List<T> ret = new LinkedList<T>();
for (Extension ex : all.values()) {
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/NamespaceHandler.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/NamespaceHandler.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/NamespaceHandler.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/NamespaceHandler.java
Thu Oct 6 16:15:26 2011
@@ -18,10 +18,15 @@
*/
package org.apache.cxf.bus.spring;
+import org.w3c.dom.Element;
+
import org.apache.cxf.configuration.spring.SimpleBeanDefinitionParser;
import org.apache.cxf.feature.FastInfosetFeature;
import org.apache.cxf.feature.LoggingFeature;
+import org.apache.cxf.workqueue.AutomaticWorkQueueImpl;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
+import org.springframework.beans.factory.xml.ParserContext;
public class NamespaceHandler extends NamespaceHandlerSupport {
public void init() {
@@ -31,5 +36,22 @@ public class NamespaceHandler extends Na
new
SimpleBeanDefinitionParser(LoggingFeature.class));
registerBeanDefinitionParser("fastinfoset",
new
SimpleBeanDefinitionParser(FastInfosetFeature.class));
+
+ registerBeanDefinitionParser("workqueue",
+ new
SimpleBeanDefinitionParser(AutomaticWorkQueueImpl.class) {
+
+ protected void processNameAttribute(Element element,
+ ParserContext ctx,
+ BeanDefinitionBuilder bean,
+ String val) {
+ bean.addPropertyValue("name", val);
+ element.removeAttribute("name");
+ if (!element.hasAttribute("id")) {
+ val = "cxf.workqueue." + val;
+ element.setAttribute("id", val);
+ }
+
+ }
+ });
}
}
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/SpringBeanLocator.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/SpringBeanLocator.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/SpringBeanLocator.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/bus/spring/SpringBeanLocator.java
Thu Oct 6 16:15:26 2011
@@ -37,6 +37,7 @@ import org.apache.cxf.configuration.Conf
import org.osgi.framework.ServiceReference;
import org.springframework.beans.Mergeable;
import org.springframework.beans.PropertyValue;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.context.ApplicationContext;
@@ -95,6 +96,18 @@ public class SpringBeanLocator implement
}
}
+ public <T> T getBeanOfType(String name, Class<T> type) {
+ T t = null;
+ try {
+ t = context.getBean(name, type);
+ } catch (NoSuchBeanDefinitionException nsbde) {
+ //ignore
+ }
+ if (t == null) {
+ t = orig.getBeanOfType(name, type);
+ }
+ return t;
+ }
/** {@inheritDoc}*/
public List<String> getBeanNamesOfType(Class<?> type) {
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/configuration/blueprint/AbstractBPBeanDefinitionParser.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/configuration/blueprint/AbstractBPBeanDefinitionParser.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/configuration/blueprint/AbstractBPBeanDefinitionParser.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/configuration/blueprint/AbstractBPBeanDefinitionParser.java
Thu Oct 6 16:15:26 2011
@@ -158,7 +158,9 @@ public abstract class AbstractBPBeanDefi
} else {
if ("depends-on".equals(name)) {
bean.addDependsOn(val);
- } else if (!"id".equals(name) && !"name".equals(name) &&
isAttribute(pre, name)) {
+ } else if ("name".equals(name)) {
+ processNameAttribute(element, ctx, bean, val);
+ } else if (!"id".equals(name) && isAttribute(pre, name)) {
if ("bus".equals(name)) {
if (this.hasBusProperty()) {
bean.addProperty("bus", getBusRef(ctx, val));
@@ -173,7 +175,12 @@ public abstract class AbstractBPBeanDefi
}
return setBus;
}
-
+ protected void processNameAttribute(Element element,
+ ParserContext ctx,
+ MutableBeanMetadata bean,
+ String val) {
+ //nothing
+ }
protected void mapAttribute(MutableBeanMetadata bean, Element e,
String name, String val, ParserContext
context) {
mapToProperty(bean, name, val, context);
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=1179688&r1=1179687&r2=1179688&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 16:15:26 2011
@@ -36,31 +36,35 @@ import java.util.concurrent.locks.Reentr
import java.util.logging.Level;
import java.util.logging.Logger;
-import javax.annotation.PostConstruct;
-import javax.annotation.Resource;
-import javax.management.JMException;
-
import org.apache.cxf.common.classloader.ClassLoaderUtils;
import org.apache.cxf.common.classloader.ClassLoaderUtils.ClassLoaderHolder;
+import org.apache.cxf.common.injection.NoJSR250Annotations;
import org.apache.cxf.common.logging.LogUtils;
-import org.apache.cxf.management.InstrumentationManager;
+import org.apache.cxf.common.util.ReflectionUtil;
-public class AutomaticWorkQueueImpl extends ThreadPoolExecutor implements
AutomaticWorkQueue {
+@NoJSR250Annotations
+public class AutomaticWorkQueueImpl implements AutomaticWorkQueue {
static final int DEFAULT_MAX_QUEUE_SIZE = 256;
private static final Logger LOG =
LogUtils.getL7dLogger(AutomaticWorkQueueImpl.class);
+
+ String name = "default";
int maxQueueSize;
- DelayQueue<DelayedTaskWrapper> delayQueue = new
DelayQueue<DelayedTaskWrapper>();
- WatchDog watchDog = new WatchDog(delayQueue);
+ int initialThreads;
+ int lowWaterMark;
+ int highWaterMark;
+ long dequeueTimeout;
+
+ ThreadPoolExecutor executor;
+ AWQThreadFactory threadFactory;
+ ReentrantLock mainLock;
+
+
+ DelayQueue<DelayedTaskWrapper> delayQueue;
+ WatchDog watchDog;
- WorkQueueManagerImpl manager;
- String name = "default";
- final int corePoolSize;
- final int maxPoolSize;
- final ReentrantLock mainLock;
-
public AutomaticWorkQueueImpl() {
this(DEFAULT_MAX_QUEUE_SIZE);
}
@@ -91,63 +95,76 @@ public class AutomaticWorkQueueImpl exte
int lowWaterMark,
long dequeueTimeout,
String name) {
-
- super(-1 == lowWaterMark ? Integer.MAX_VALUE : lowWaterMark,
- -1 == highWaterMark ? Integer.MAX_VALUE : highWaterMark,
- TimeUnit.MILLISECONDS.toMillis(dequeueTimeout),
TimeUnit.MILLISECONDS,
- mqs == -1 ? new
LinkedBlockingQueue<Runnable>(DEFAULT_MAX_QUEUE_SIZE)
- : new LinkedBlockingQueue<Runnable>(mqs),
- createThreadFactory(name));
-
- maxQueueSize = mqs == -1 ? DEFAULT_MAX_QUEUE_SIZE : mqs;
-
-
- lowWaterMark = -1 == lowWaterMark ? Integer.MAX_VALUE : lowWaterMark;
- highWaterMark = -1 == highWaterMark ? Integer.MAX_VALUE :
highWaterMark;
-
- StringBuilder buf = new StringBuilder();
- buf.append("Constructing automatic work queue with:\n");
- buf.append("max queue size: " + maxQueueSize + "\n");
- buf.append("initialThreads: " + initialThreads + "\n");
- buf.append("lowWaterMark: " + lowWaterMark + "\n");
- buf.append("highWaterMark: " + highWaterMark + "\n");
- LOG.fine(buf.toString());
-
- if (initialThreads > highWaterMark) {
- initialThreads = highWaterMark;
- }
-
- // as we cannot prestart more core than corePoolSize initial threads,
we temporarily
- // change the corePoolSize to the number of initial threads
- // this is important as otherwise these threads will be created only
when the queue has filled up,
- // potentially causing problems with starting up under heavy load
- if (initialThreads < Integer.MAX_VALUE && initialThreads > 0) {
- setCorePoolSize(initialThreads);
- int started = prestartAllCoreThreads();
- if (started < initialThreads) {
- LOG.log(Level.WARNING, "THREAD_START_FAILURE_MSG", new
Object[] {started, initialThreads});
+ this.maxQueueSize = mqs == -1 ? DEFAULT_MAX_QUEUE_SIZE : mqs;
+ this.initialThreads = initialThreads;
+ this.highWaterMark = -1 == highWaterMark ? Integer.MAX_VALUE :
highWaterMark;
+ this.lowWaterMark = -1 == lowWaterMark ? Integer.MAX_VALUE :
lowWaterMark;
+ this.dequeueTimeout = dequeueTimeout;
+ this.name = name;
+ }
+
+ protected synchronized ThreadPoolExecutor getExecutor() {
+ if (executor == null) {
+ threadFactory = createThreadFactory(name);
+ executor = new ThreadPoolExecutor(lowWaterMark,
+ highWaterMark,
+
TimeUnit.MILLISECONDS.toMillis(dequeueTimeout),
+ TimeUnit.MILLISECONDS,
+ new
LinkedBlockingQueue<Runnable>(maxQueueSize),
+ threadFactory) {
+ @Override
+ protected void terminated() {
+ ThreadFactory f = executor.getThreadFactory();
+ if (f instanceof AWQThreadFactory) {
+ ((AWQThreadFactory)f).shutdown();
+ }
+ if (watchDog != null) {
+ watchDog.shutdown();
+ }
+ }
+ };
+
+
+ StringBuilder buf = new StringBuilder();
+ buf.append("Constructing automatic work queue with:\n");
+ buf.append("max queue size: " + maxQueueSize + "\n");
+ buf.append("initialThreads: " + initialThreads + "\n");
+ buf.append("lowWaterMark: " + lowWaterMark + "\n");
+ buf.append("highWaterMark: " + highWaterMark + "\n");
+ LOG.fine(buf.toString());
+
+ if (initialThreads > highWaterMark) {
+ initialThreads = highWaterMark;
}
- setCorePoolSize(lowWaterMark);
- }
-
- // start the watch dog thread
- watchDog.setDaemon(true);
- watchDog.start();
-
- corePoolSize = this.getCorePoolSize();
- maxPoolSize = this.getMaximumPoolSize();
-
- ReentrantLock l = null;
- try {
- Field f = ThreadPoolExecutor.class.getDeclaredField("mainLock");
- f.setAccessible(true);
- l = (ReentrantLock)f.get(this);
- } catch (Throwable t) {
- l = new ReentrantLock();
+
+ // as we cannot prestart more core than corePoolSize initial
threads, we temporarily
+ // change the corePoolSize to the number of initial threads
+ // this is important as otherwise these threads will be created
only when
+ // the queue has filled up,
+ // potentially causing problems with starting up under heavy load
+ if (initialThreads < Integer.MAX_VALUE && initialThreads > 0) {
+ executor.setCorePoolSize(initialThreads);
+ int started = executor.prestartAllCoreThreads();
+ if (started < initialThreads) {
+ LOG.log(Level.WARNING, "THREAD_START_FAILURE_MSG",
+ new Object[] {started, initialThreads});
+ }
+ executor.setCorePoolSize(lowWaterMark);
+ }
+
+ ReentrantLock l = null;
+ try {
+ Field f =
ThreadPoolExecutor.class.getDeclaredField("mainLock");
+ ReflectionUtil.setAccessible(f);
+ l = (ReentrantLock)f.get(executor);
+ } catch (Throwable t) {
+ l = new ReentrantLock();
+ }
+ mainLock = l;
}
- mainLock = l;
+ return executor;
}
- private static ThreadFactory createThreadFactory(final String name) {
+ private static AWQThreadFactory createThreadFactory(final String name) {
ThreadGroup group;
try {
//Try and find the highest level ThreadGroup that we're allowed to
use.
@@ -294,40 +311,16 @@ public class AutomaticWorkQueueImpl exte
}
}
- @Resource(name = "org.apache.cxf.workqueue.WorkQueueManager")
- public void setManager(WorkQueueManager mgr) {
- manager = (WorkQueueManagerImpl)mgr;
- }
- public WorkQueueManager getManager() {
- return manager;
- }
-
public void setName(String s) {
name = s;
- ThreadFactory factory = this.getThreadFactory();
- if (factory instanceof AWQThreadFactory) {
- ((AWQThreadFactory)factory).setName(s);
+ if (threadFactory != null) {
+ threadFactory.setName(s);
}
}
public String getName() {
return name;
}
-
- @PostConstruct
- public void register() {
- if (manager != null) {
- manager.addNamedWorkQueue(name, this);
- InstrumentationManager imanager =
manager.getBus().getExtension(InstrumentationManager.class);
- if (null != imanager) {
- try {
- imanager.register(new WorkQueueImplMBeanWrapper(this));
- } catch (JMException jmex) {
- LOG.log(Level.WARNING , jmex.getMessage(), jmex);
- }
- }
- }
- }
-
+
public String toString() {
StringBuilder buf = new StringBuilder();
buf.append(super.toString());
@@ -371,21 +364,21 @@ public class AutomaticWorkQueueImpl exte
//of threads. Thus, we'll set the core size to the max,
//add the runnable, and set back. That will cause the
//threads to be created as needed.
- super.execute(r);
- if (!getQueue().isEmpty() && this.getPoolSize() < maxPoolSize) {
+ ThreadPoolExecutor ex = getExecutor();
+ ex.execute(r);
+ if (!ex.getQueue().isEmpty() && this.getPoolSize() < highWaterMark) {
mainLock.lock();
try {
int ps = this.getPoolSize();
- int sz = getQueue().size();
+ int sz = executor.getQueue().size();
int sz2 = this.getActiveCount();
if ((sz + sz2) > ps) {
Method m =
ThreadPoolExecutor.class.getDeclaredMethod("addIfUnderMaximumPoolSize",
Runnable.class);
- m.setAccessible(true);
- m.invoke(this, new Object[1]);
+ ReflectionUtil.setAccessible(m).invoke(this, new
Object[1]);
}
- } catch (Exception ex) {
+ } catch (Exception exc) {
//ignore
} finally {
mainLock.unlock();
@@ -399,34 +392,35 @@ public class AutomaticWorkQueueImpl exte
execute(work);
} catch (RejectedExecutionException ree) {
try {
- getQueue().offer(work, timeout, TimeUnit.MILLISECONDS);
+ getExecutor().getQueue().offer(work, timeout,
TimeUnit.MILLISECONDS);
} catch (InterruptedException ie) {
throw new RejectedExecutionException(ie);
}
}
}
- public void schedule(final Runnable work, final long delay) {
+ public synchronized void schedule(final Runnable work, final long delay) {
+ if (delayQueue == null) {
+ delayQueue = new DelayQueue<DelayedTaskWrapper>();
+ watchDog = new WatchDog(delayQueue);
+ watchDog.setDaemon(true);
+ watchDog.start();
+ }
delayQueue.put(new DelayedTaskWrapper(work, delay));
}
// AutomaticWorkQueue interface
public void shutdown(boolean processRemainingWorkItems) {
- if (!processRemainingWorkItems) {
- getQueue().clear();
- }
- shutdown();
- }
- @Override
- protected void terminated() {
- ThreadFactory f = this.getThreadFactory();
- if (f instanceof AWQThreadFactory) {
- ((AWQThreadFactory)f).shutdown();
+ if (executor != null) {
+ if (!processRemainingWorkItems) {
+ executor.getQueue().clear();
+ }
+ executor.shutdown();
}
- watchDog.shutdown();
}
+
/**
* Gets the maximum size (capacity) of the backing queue.
* @return the maximum size (capacity) of the backing queue.
@@ -440,33 +434,76 @@ public class AutomaticWorkQueueImpl exte
* @return the current size of the backing queue.
*/
public long getSize() {
- return getQueue().size();
+ return executor == null ? 0 : executor.getQueue().size();
}
public boolean isEmpty() {
- return getQueue().size() == 0;
+ return executor == null ? true : executor.getQueue().size() == 0;
}
boolean isFull() {
- return getQueue().remainingCapacity() == 0;
+ return executor == null ? false :
executor.getQueue().remainingCapacity() == 0;
}
public int getHighWaterMark() {
- int hwm = getMaximumPoolSize();
+ int hwm = executor == null ? highWaterMark :
executor.getMaximumPoolSize();
return hwm == Integer.MAX_VALUE ? -1 : hwm;
}
public int getLowWaterMark() {
- int lwm = getCorePoolSize();
+ int lwm = executor == null ? lowWaterMark : executor.getCorePoolSize();
return lwm == Integer.MAX_VALUE ? -1 : lwm;
}
public void setHighWaterMark(int hwm) {
- setMaximumPoolSize(hwm < 0 ? Integer.MAX_VALUE : hwm);
+ highWaterMark = hwm < 0 ? Integer.MAX_VALUE : hwm;
+ if (executor != null) {
+ executor.setMaximumPoolSize(highWaterMark);
+ }
}
public void setLowWaterMark(int lwm) {
- setCorePoolSize(lwm < 0 ? 0 : lwm);
+ lowWaterMark = lwm < 0 ? 0 : lwm;
+ if (executor != null) {
+ executor.setCorePoolSize(lowWaterMark);
+ }
+ }
+
+ public void setInitialSize(int initialSize) {
+ this.initialThreads = initialSize;
+ }
+
+ public void setQueueSize(int size) {
+ this.maxQueueSize = size;
+ }
+
+ public void setDequeueTimeout(long l) {
+ this.dequeueTimeout = l;
+ }
+
+ public boolean isShutdown() {
+ if (executor == null) {
+ return false;
+ }
+ return executor.isShutdown();
+ }
+ public int getLargestPoolSize() {
+ if (executor == null) {
+ return 0;
+ }
+ return executor.getLargestPoolSize();
+ }
+ public int getPoolSize() {
+ if (executor == null) {
+ return 0;
+ }
+ return executor.getPoolSize();
+ }
+ public int getActiveCount() {
+ if (executor == null) {
+ return 0;
+ }
+ return executor.getActiveCount();
}
}
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueImplMBeanWrapper.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueImplMBeanWrapper.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueImplMBeanWrapper.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueImplMBeanWrapper.java
Thu Oct 6 16:15:26 2011
@@ -35,9 +35,12 @@ public class WorkQueueImplMBeanWrapper i
private static final String TYPE_VALUE = "WorkQueues";
private AutomaticWorkQueueImpl aWorkQueue;
+ private WorkQueueManager manager;
- public WorkQueueImplMBeanWrapper(AutomaticWorkQueueImpl wq) {
- aWorkQueue = wq;
+ public WorkQueueImplMBeanWrapper(AutomaticWorkQueueImpl wq,
+ WorkQueueManager mgr) {
+ aWorkQueue = wq;
+ manager = mgr;
}
@ManagedAttribute(description = "The WorkQueueMaxSize",
@@ -98,10 +101,9 @@ public class WorkQueueImplMBeanWrapper i
}
public ObjectName getObjectName() throws JMException {
- WorkQueueManager mgr = aWorkQueue.getManager();
String busId = "cxf";
- if (mgr instanceof WorkQueueManagerImpl) {
- busId = ((WorkQueueManagerImpl)mgr).getBus().getId();
+ if (manager instanceof WorkQueueManagerImpl) {
+ busId = ((WorkQueueManagerImpl)manager).getBus().getId();
}
StringBuilder buffer = new StringBuilder();
buffer.append(ManagementConstants.DEFAULT_DOMAIN_NAME + ":");
Modified:
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueManagerImpl.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueManagerImpl.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueManagerImpl.java
(original)
+++
cxf/trunk/rt/core/src/main/java/org/apache/cxf/workqueue/WorkQueueManagerImpl.java
Thu Oct 6 16:15:26 2011
@@ -19,6 +19,7 @@
package org.apache.cxf.workqueue;
+import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
@@ -30,6 +31,7 @@ import javax.management.JMException;
import org.apache.cxf.Bus;
import org.apache.cxf.common.injection.NoJSR250Annotations;
import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.configuration.ConfiguredBeanLocator;
import org.apache.cxf.management.InstrumentationManager;
@NoJSR250Annotations(unlessNull = "bus")
@@ -40,7 +42,9 @@ public class WorkQueueManagerImpl implem
Map<String, AutomaticWorkQueue> namedQueues
= new ConcurrentHashMap<String, AutomaticWorkQueue>();
+
boolean inShutdown;
+ InstrumentationManager imanager;
Bus bus;
public WorkQueueManagerImpl() {
@@ -59,14 +63,30 @@ public class WorkQueueManagerImpl implem
this.bus = bus;
if (null != bus) {
bus.setExtension(this, WorkQueueManager.class);
- InstrumentationManager manager =
bus.getExtension(InstrumentationManager.class);
- if (null != manager) {
+ imanager = bus.getExtension(InstrumentationManager.class);
+ if (null != imanager) {
try {
- manager.register(new
WorkQueueManagerImplMBeanWrapper(this));
+ imanager.register(new
WorkQueueManagerImplMBeanWrapper(this));
} catch (JMException jmex) {
LOG.log(Level.WARNING , jmex.getMessage(), jmex);
}
}
+ ConfiguredBeanLocator locator =
bus.getExtension(ConfiguredBeanLocator.class);
+ Collection<? extends AutomaticWorkQueue> q = locator
+ .getBeansOfType(AutomaticWorkQueue.class);
+ if (q != null) {
+ for (AutomaticWorkQueue awq : q) {
+ addNamedWorkQueue(awq.getName(), awq);
+ }
+ }
+
+ if (!namedQueues.containsKey("default")) {
+ AutomaticWorkQueue defaultQueue
+ = locator.getBeanOfType("cxf.default.workqueue",
AutomaticWorkQueue.class);
+ if (defaultQueue != null) {
+ addNamedWorkQueue("default", defaultQueue);
+ }
+ }
}
}
@@ -117,15 +137,21 @@ public class WorkQueueManagerImpl implem
public AutomaticWorkQueue getNamedWorkQueue(String name) {
return namedQueues.get(name);
}
- public void addNamedWorkQueue(String name, AutomaticWorkQueue q) {
+ public final void addNamedWorkQueue(String name, AutomaticWorkQueue q) {
namedQueues.put(name, q);
+ if (imanager != null && q instanceof AutomaticWorkQueueImpl) {
+ try {
+ imanager.register(new
WorkQueueImplMBeanWrapper((AutomaticWorkQueueImpl)q, this));
+ } catch (JMException jmex) {
+ LOG.log(Level.WARNING , jmex.getMessage(), jmex);
+ }
+ }
}
private AutomaticWorkQueue createAutomaticWorkQueue() {
- AutomaticWorkQueueImpl impl = new AutomaticWorkQueueImpl("default");
- impl.setManager(this);
- impl.register();
- return impl;
+ AutomaticWorkQueue q = new AutomaticWorkQueueImpl("default");
+ addNamedWorkQueue("default", q);
+ return q;
}
}
Modified: cxf/trunk/rt/core/src/main/resources/schemas/blueprint/core.xsd
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/resources/schemas/blueprint/core.xsd?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
--- cxf/trunk/rt/core/src/main/resources/schemas/blueprint/core.xsd (original)
+++ cxf/trunk/rt/core/src/main/resources/schemas/blueprint/core.xsd Thu Oct 6
16:15:26 2011
@@ -21,13 +21,17 @@
xmlns:beans="http://www.osgi.org/xmlns/blueprint/v1.0.0";
xmlns:cxf-beans="http://cxf.apache.org/configuration/beans";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xmlns:ptp="http://cxf.apache.org/configuration/parameterized-types";
+
targetNamespace="http://cxf.apache.org/blueprint/core";
elementFormDefault="qualified"
attributeFormDefault="unqualified" >
<xsd:import namespace="http://www.osgi.org/xmlns/blueprint/v1.0.0";
schemaLocation="http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd"/>
<xsd:import namespace="http://cxf.apache.org/configuration/beans";
schemaLocation="http://cxf.apache.org/schemas/configuration/cxf-beans.xsd"/>
-
+ <xsd:import
namespace="http://cxf.apache.org/configuration/parameterized-types";
+
schemaLocation="http://cxf.apache.org/schemas/configuration/parameterized-types.xsd"/>
+
<xsd:annotation>
<xsd:documentation>
This schema defines beans representing the CXF bus and its core features.
@@ -61,6 +65,22 @@
<xsd:attribute name="force" type="xsd:boolean" use="optional"
default="false"/>
</xsd:complexType>
</xsd:element>
+ <xsd:element name="workqueue">
+ <xsd:annotation>
+ <xsd:documentation>
+ Configuration for workqueues that CXF uses
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence />
+ <xsd:attribute name="highWaterMark" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="lowWaterMark" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="initialSize" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="queueSize" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="name" type="xsd:string" use="required"/>
+ <xsd:attribute name="dequeueTimeout" type="ptp:ParameterizedLong"
use="optional"/>
+ </xsd:complexType>
+ </xsd:element>
<xsd:element name="bus">
<xsd:complexType>
Modified: cxf/trunk/rt/core/src/main/resources/schemas/core.xsd
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/main/resources/schemas/core.xsd?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
--- cxf/trunk/rt/core/src/main/resources/schemas/core.xsd (original)
+++ cxf/trunk/rt/core/src/main/resources/schemas/core.xsd Thu Oct 6 16:15:26
2011
@@ -21,13 +21,17 @@
xmlns:beans="http://www.springframework.org/schema/beans";
xmlns:cxf-beans="http://cxf.apache.org/configuration/beans";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xmlns:ptp="http://cxf.apache.org/configuration/parameterized-types";
+
targetNamespace="http://cxf.apache.org/core";
elementFormDefault="qualified"
attributeFormDefault="unqualified" >
<xsd:import namespace="http://www.springframework.org/schema/beans";
schemaLocation="http://www.springframework.org/schema/beans/spring-beans.xsd"/>
<xsd:import namespace="http://cxf.apache.org/configuration/beans";
schemaLocation="http://cxf.apache.org/schemas/configuration/cxf-beans.xsd"/>
-
+ <xsd:import
namespace="http://cxf.apache.org/configuration/parameterized-types";
+
schemaLocation="http://cxf.apache.org/schemas/configuration/parameterized-types.xsd"/>
+
<xsd:annotation>
<xsd:documentation>
This schema defines beans representing the CXF bus and its core features.
@@ -61,6 +65,22 @@
<xsd:attribute name="force" type="xsd:boolean" use="optional"
default="false"/>
</xsd:complexType>
</xsd:element>
+ <xsd:element name="workqueue">
+ <xsd:annotation>
+ <xsd:documentation>
+ Configuration for workqueues that CXF uses
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence />
+ <xsd:attribute name="highWaterMark" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="lowWaterMark" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="initialSize" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="queueSize" type="ptp:ParameterizedInt"
use="optional"/>
+ <xsd:attribute name="name" type="xsd:string" use="required"/>
+ <xsd:attribute name="dequeueTimeout" type="ptp:ParameterizedLong"
use="optional"/>
+ </xsd:complexType>
+ </xsd:element>
<xsd:element name="bus">
<xsd:complexType>
Modified:
cxf/trunk/rt/core/src/test/java/org/apache/cxf/workqueue/AutomaticWorkQueueTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/core/src/test/java/org/apache/cxf/workqueue/AutomaticWorkQueueTest.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/core/src/test/java/org/apache/cxf/workqueue/AutomaticWorkQueueTest.java
(original)
+++
cxf/trunk/rt/core/src/test/java/org/apache/cxf/workqueue/AutomaticWorkQueueTest.java
Thu Oct 6 16:15:26 2011
@@ -91,13 +91,14 @@ public class AutomaticWorkQueueTest exte
// We haven't enqueued anything yet, so should be zero
assertEquals(0, workqueue.getSize());
- assertEquals(INITIAL_SIZE, workqueue.getPoolSize());
+ assertEquals(0, workqueue.getPoolSize());
// Check that no threads are working yet, as we haven't enqueued
// anything yet.
assertEquals(0, workqueue.getActiveCount());
workqueue.execute(new TestWorkItem(), TIMEOUT);
+ assertEquals(INITIAL_SIZE, workqueue.getPoolSize());
// Give threads a chance to dequeue (5sec max)
int i = 0;
@@ -129,7 +130,7 @@ public class AutomaticWorkQueueTest exte
// initial size and no threads should be working
//
assertEquals(0, workqueue.getSize());
- assertEquals(INITIAL_SIZE, workqueue.getPoolSize());
+ assertEquals(0, workqueue.getPoolSize());
assertEquals(0, workqueue.getActiveCount());
BlockingWorkItem[] workItems = new
BlockingWorkItem[DEFAULT_HIGH_WATER_MARK];
Modified:
cxf/trunk/rt/management/src/test/java/org/apache/cxf/management/InstrumentationManagerTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/management/src/test/java/org/apache/cxf/management/InstrumentationManagerTest.java?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
---
cxf/trunk/rt/management/src/test/java/org/apache/cxf/management/InstrumentationManagerTest.java
(original)
+++
cxf/trunk/rt/management/src/test/java/org/apache/cxf/management/InstrumentationManagerTest.java
Thu Oct 6 16:15:26 2011
@@ -73,13 +73,20 @@ public class InstrumentationManagerTest
ObjectName name = new
ObjectName(ManagementConstants.DEFAULT_DOMAIN_NAME
+ ":type=WorkQueues,*");
Set s = mbs.queryNames(name, null);
- assertTrue(s.size() == 1);
+ assertEquals(2, s.size());
Iterator it = s.iterator();
while (it.hasNext()) {
- ObjectName n = (ObjectName)it.next();
+ ObjectName n = (ObjectName)it.next();
Long result =
- (Long)mbs.invoke(n, "getWorkQueueMaxSize", new Object[0], new
String[0]);
+ (Long)mbs.invoke(n, "getWorkQueueMaxSize", new Object[0], new
String[0]);
assertEquals(result, Long.valueOf(256));
+ Integer hwm =
+ (Integer)mbs.invoke(n, "getHighWaterMark", new Object[0], new
String[0]);
+ if (n.getCanonicalName().contains("test-wq")) {
+ assertEquals(10, hwm.intValue());
+ } else {
+ assertEquals(15, hwm.intValue());
+ }
}
bus.shutdown(true);
Modified: cxf/trunk/rt/management/src/test/resources/managed-spring.xml
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/management/src/test/resources/managed-spring.xml?rev=1179688&r1=1179687&r2=1179688&view=diff
==============================================================================
--- cxf/trunk/rt/management/src/test/resources/managed-spring.xml (original)
+++ cxf/trunk/rt/management/src/test/resources/managed-spring.xml Thu Oct 6
16:15:26 2011
@@ -35,4 +35,6 @@
<property name="JMXServiceURL"
value="service:jmx:rmi:///jndi/rmi://localhost:9914/jmxrmi" />
</bean>
+ <cxf:workqueue name="default" highWaterMark="15" lowWaterMark="5"/>
+ <cxf:workqueue name="test-wq" highWaterMark="10" lowWaterMark="2"/>
</beans>
\ No newline at end of file
