Merge pull request #9405 from jmesnil/WFLY-7648_global_client_thread_…

…pool_size

[WFLY-7648] Configuration of ActiveMQ client thread pools

messaging-activemq/src/main/java/org/wildfly/extension/messaging/activemq/MessagingSubsystemAdd.java

@@ -22,6 +22,10 @@
 
 package org.wildfly.extension.messaging.activemq;
 
+import static org.wildfly.extension.messaging.activemq.MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE;
+import static org.wildfly.extension.messaging.activemq.MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE;
+
+import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
 import org.jboss.as.controller.AbstractBoottimeAddStepHandler;
 import org.jboss.as.controller.OperationContext;
 import org.jboss.as.controller.OperationFailedException;
@@ -50,11 +54,7 @@ class MessagingSubsystemAdd extends AbstractBoottimeAddStepHandler {
     public static final MessagingSubsystemAdd INSTANCE = new MessagingSubsystemAdd();
 
     private MessagingSubsystemAdd() {
-    }
-
-    @Override
-    protected void populateModel(ModelNode operation, ModelNode model) throws OperationFailedException {
-        model.setEmptyObject();
+        super(MessagingSubsystemRootResourceDefinition.ATTRIBUTES);
     }
 
     @Override
@@ -75,6 +75,9 @@ protected void execute(DeploymentProcessorTarget processorTarget) {
                 processorTarget.addDeploymentProcessor(MessagingExtension.SUBSYSTEM_NAME, Phase.INSTALL, Phase.INSTALL_MESSAGING_XML_RESOURCES, new MessagingXmlInstallDeploymentUnitProcessor());
             }
         }, OperationContext.Stage.RUNTIME);
-    }
 
+        int threadPoolMaxSize = GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE.resolveModelAttribute(context, operation).asInt();
+        int scheduledThreadPoolMaxSize = GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE.resolveModelAttribute(context, operation).asInt();
+        ActiveMQClient.setGlobalThreadPoolProperties(threadPoolMaxSize, scheduledThreadPoolMaxSize);
+    }
 }

messaging-activemq/src/main/java/org/wildfly/extension/messaging/activemq/MessagingSubsystemParser_1_1.java

@@ -72,7 +72,10 @@ public class MessagingSubsystemParser_1_1 implements XMLStreamConstants, XMLElem
     private static final PersistentResourceXMLDescription xmlDescription;
 
     static {
-        xmlDescription = builder(MessagingSubsystemRootResourceDefinition.INSTANCE)
+        xmlDescription = builder(MessagingExtension.SUBSYSTEM_PATH)
+                .addAttributes(
+                        MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE,
+                        MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE)
                 .addChild(
                         builder(ServerDefinition.INSTANCE)
                                 .addAttributes(

messaging-activemq/src/main/java/org/wildfly/extension/messaging/activemq/MessagingSubsystemRootResourceDefinition.java

@@ -22,20 +22,25 @@
 
 package org.wildfly.extension.messaging.activemq;
 
+import static org.jboss.as.controller.SimpleAttributeDefinitionBuilder.create;
 import static org.jboss.as.controller.transform.description.RejectAttributeChecker.DEFINED;
+import static org.jboss.dmr.ModelType.INT;
 
+import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 
+import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
 import org.jboss.as.controller.AttributeDefinition;
 import org.jboss.as.controller.PersistentResourceDefinition;
 import org.jboss.as.controller.ReloadRequiredRemoveStepHandler;
+import org.jboss.as.controller.SimpleAttributeDefinition;
 import org.jboss.as.controller.SubsystemRegistration;
 import org.jboss.as.controller.transform.description.AttributeConverter.DefaultValueAttributeConverter;
 import org.jboss.as.controller.transform.description.DiscardAttributeChecker;
 import org.jboss.as.controller.transform.description.ResourceTransformationDescriptionBuilder;
 import org.jboss.as.controller.transform.description.TransformationDescription;
 import org.jboss.as.controller.transform.description.TransformationDescriptionBuilder;
+import org.jboss.dmr.ModelNode;
 import org.wildfly.extension.messaging.activemq.jms.ConnectionFactoryAttributes;
 
 /**
@@ -45,6 +50,29 @@
  */
 public class MessagingSubsystemRootResourceDefinition extends PersistentResourceDefinition {
 
+    public static final SimpleAttributeDefinition GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE = create("global-client-thread-pool-max-size", INT)
+            .setAttributeGroup("global-client")
+            .setXmlName("thread-pool-max-size")
+            .setAllowNull(true)
+            .setDefaultValue(new ModelNode(ActiveMQClient.DEFAULT_THREAD_POOL_MAX_SIZE))
+            .setAllowExpression(true)
+            .setRestartAllServices()
+            .build();
+
+    public static final SimpleAttributeDefinition GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE = create("global-client-scheduled-thread-pool-max-size", INT)
+            .setAttributeGroup("global-client")
+            .setXmlName("scheduled-thread-pool-max-size")
+            .setAllowNull(true)
+            .setDefaultValue(new ModelNode(ActiveMQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE))
+            .setAllowExpression(true)
+            .setRestartAllServices()
+            .build();
+
+    public static final AttributeDefinition[] ATTRIBUTES = {
+            GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE,
+            GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE
+    };
+
     public static final MessagingSubsystemRootResourceDefinition INSTANCE = new MessagingSubsystemRootResourceDefinition();
 
     private MessagingSubsystemRootResourceDefinition() {
@@ -56,17 +84,20 @@ private MessagingSubsystemRootResourceDefinition() {
 
     @Override
     public Collection<AttributeDefinition> getAttributes() {
-        return Collections.emptyList();
+        return Arrays.asList(ATTRIBUTES);
     }
 
     public static void registerTransformers(SubsystemRegistration subsystemRegistration) {
         registerTransformers_EAP_7_0_0(subsystemRegistration);
     }
 
     private static void registerTransformers_EAP_7_0_0(SubsystemRegistration subsystemRegistration) {
-        final ResourceTransformationDescriptionBuilder builder = TransformationDescriptionBuilder.Factory.createSubsystemInstance();
+        final ResourceTransformationDescriptionBuilder subsystem = TransformationDescriptionBuilder.Factory.createSubsystemInstance();
+
+        rejectDefinedAttributeWithDefaultValue(subsystem, MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE,
+                MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE);
 
-        ResourceTransformationDescriptionBuilder server = builder.addChildResource(MessagingExtension.SERVER_PATH);
+        ResourceTransformationDescriptionBuilder server = subsystem.addChildResource(MessagingExtension.SERVER_PATH);
         // reject journal-datasource, journal-bindings-table introduced in management version 2.0.0 if it is defined and different from the default value.
         rejectDefinedAttributeWithDefaultValue(server, ServerDefinition.JOURNAL_DATASOURCE,
                 ServerDefinition.JOURNAL_MESSAGES_TABLE,
@@ -89,7 +120,7 @@ private static void registerTransformers_EAP_7_0_0(SubsystemRegistration subsyst
         // reject min-pool-size whose default value has been changed in  management version 2.0.0
         defaultValueAttributeConverter(pooledConnectionFactory, ConnectionFactoryAttributes.Pooled.MIN_POOL_SIZE);
 
-        TransformationDescription.Tools.register(builder.build(), subsystemRegistration, MessagingExtension.VERSION_1_0_0);
+        TransformationDescription.Tools.register(subsystem.build(), subsystemRegistration, MessagingExtension.VERSION_1_0_0);
     }
 
     /**

messaging-activemq/src/main/resources/org/wildfly/extension/messaging/activemq/LocalDescriptions.properties

@@ -260,6 +260,8 @@ ha-policy.scale-down-group-name=Name of the group used to scale down.
 ha-policy.scale-down=Configure whether this server send its messages to another live server in the scale-down cluster when it is shutdown cleanly.
 ha-policy=A messaging resource that allows you to configure High Availability for the ActiveMQ server (the value of ha-policy can be live-only, replication-master, replication-slave, or replication-colocated).
 legacy-connection-factory=Connection Factory resource used by legacy HornetQ clients to connect to the messaging-activemq subsystem.
+messaging-activemq.global-client-scheduled-thread-pool-max-size=Maximum size of the pool of threads used by all ActiveMQ clients running inside this server.
+messaging-activemq.global-client-thread-pool-max-size=Maximum size of the pool of scheduled threads used by all ActiveMQ clients running inside this server.
 server.active=Whether the server is active (and accepting connections) or passive (in backup mode, waiting for failover).
 server.add=Operation adding a ActiveMQ server.
 server.allow-failback=Whether this server will automatically shutdown if the original live server comes back up.

messaging-activemq/src/main/resources/schema/wildfly-messaging-activemq_1_1.xsd

@@ -32,6 +32,12 @@
     <xs:element name="subsystem">
         <xs:complexType>
             <xs:sequence>
+                <xs:element maxOccurs="1" minOccurs="0" name="global-client">
+                    <xs:complexType>
+                        <xs:attribute name="thread-pool-max-size" type="xs:int" />
+                        <xs:attribute name="scheduled-thread-pool-max-size" type="xs:int" />
+                    </xs:complexType>
+                </xs:element>
                 <xs:element maxOccurs="unbounded" minOccurs="0" name="server" type="serverType" />
                 <xs:element maxOccurs="unbounded" minOccurs="0" name="jms-bridge" type="jms-bridgeType" />
             </xs:sequence>

messaging-activemq/src/test/java/org/wildfly/extension/messaging/activemq/MessagingActiveMQSubsystem_1_1_TestCase.java

@@ -141,6 +141,10 @@ private void testRejectingTransformers(ModelTestControllerVersion controllerVers
         System.out.println("ops = " + ops);
         PathAddress subsystemAddress = PathAddress.pathAddress(SUBSYSTEM_PATH);
         ModelTestUtils.checkFailedTransformedBootOperations(mainServices, messagingVersion, ops, new FailedOperationTransformationConfig()
+                .addFailedAttribute(subsystemAddress,
+                        new FailedOperationTransformationConfig.NewAttributesConfig(
+                                MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_THREAD_POOL_MAX_SIZE,
+                                MessagingSubsystemRootResourceDefinition.GLOBAL_CLIENT_SCHEDULED_THREAD_POOL_MAX_SIZE))
                 .addFailedAttribute(subsystemAddress.append(SERVER_PATH),
                         new FailedOperationTransformationConfig.NewAttributesConfig(
                                 ServerDefinition.JOURNAL_DATASOURCE,

messaging-activemq/src/test/resources/org/wildfly/extension/messaging/activemq/subsystem_1_1.xml

@@ -1,4 +1,6 @@
 <subsystem xmlns="urn:jboss:domain:messaging-activemq:1.1">
+    <global-client thread-pool-max-size="${global.client.thread-pool-max-size:32}"
+           scheduled-thread-pool-max-size="${global.client.scheduled.thread-pool-max-size:54}" />
     <server name="default"
             persistence-enabled="${persistence.enabled:false}"
             persist-id-cache="${persist.id.cache:false}"

messaging-activemq/src/test/resources/org/wildfly/extension/messaging/activemq/subsystem_1_1_reject_transform.xml

@@ -1,4 +1,6 @@
 <subsystem xmlns="urn:jboss:domain:messaging-activemq:1.1">
+    <global-client thread-pool-max-size="${global.client.thread-pool-max-size:32}"
+                   scheduled-thread-pool-max-size="${global.client.scheduled.thread-pool-max-size:54}" />
     <server name="default">
         <journal datasource="fooDS"
                  messages-table="MY_MESSAGES"